{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "91b4a911-7650-444a-85f1-1310fd55544c",
   "metadata": {},
   "source": [
    "# CNN和RNN"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "64ad9e56-e119-4b68-ae59-c0e6879fc462",
   "metadata": {},
   "source": [
    "## CNN示例"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c5513768-5a18-4799-9abf-3cb5c562c51b",
   "metadata": {},
   "source": [
    "### 卷积操作"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4479008c-63e8-42c2-83f6-d2d7f7525715",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 定义卷积层\n",
    "conv_layer = nn.Conv2d(in_channels=1, out_channels=16, kernel_size=3, stride=1, padding=1)\n",
    "\n",
    "# 输入：1张28x28灰度图像，形状为(1, 1, 28, 28)\n",
    "input_image = torch.randn(1, 1, 90, 90)\n",
    "\n",
    "# 前向传播\n",
    "output = conv_layer(input_image)\n",
    "\n",
    "# 输出形状：(1, 16, 90, 90)，16张28x28特征图\n",
    "print(f\"示例1(3x3卷积核，填充为1，步长为1)：{output.shape}\")\n",
    "\n",
    "# 示例2：步长的影响\n",
    "conv_layer = nn.Conv2d(in_channels=1, out_channels=16, kernel_size=3, stride=2, padding=1)\n",
    "output = conv_layer(input_image)\n",
    "print(f\"示例2(3x3卷积核，填充为1，步长为2)：{output.shape}\")\n",
    "\n",
    "# 示例3：卷积核的影响\n",
    "conv_layer = nn.Conv2d(in_channels=1, out_channels=16, kernel_size=5, stride=1, padding=1)\n",
    "output = conv_layer(input_image)\n",
    "print(f\"示例3(5x5卷积核，填充为1，步长为1)：{output.shape}\")\n",
    "\n",
    "# 示例4：卷积核和步长的影响\n",
    "conv_layer = nn.Conv2d(in_channels=1, out_channels=16, kernel_size=5, stride=2, padding=1)\n",
    "output = conv_layer(input_image)\n",
    "print(f\"示例4(5x5卷积核，填充为1，步长为2)：{output.shape}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed6eba8d-b4b5-49f7-aa94-b49c7bbc904e",
   "metadata": {},
   "source": [
    "### 池化操作示例"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0bb4a4bb-8bdb-4986-a104-e5261719b49e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 定义最大池化层\n",
    "max_pool_layer = nn.MaxPool2d(kernel_size=2, stride=2)\n",
    "avg_pool_layer = nn.AvgPool2d(kernel_size=2, stride=2)\n",
    "\n",
    "# 输入：1张8x8特征图，4个通道，形状为(1, 4, 8, 8)\n",
    "input_feature = torch.randn(1, 4, 8, 8)\n",
    "\n",
    "# 前向传播\n",
    "max_pool_output = max_pool_layer(input_feature)\n",
    "avg_pool_output = avg_pool_layer(input_feature)\n",
    "\n",
    "# 输出形状：(1, 4, 8, 8)，通道数不变，空间尺寸减半\n",
    "print(f\"示例1(2x2最大化池化核，填充为0，步长为2)：{max_pool_output.shape}\")\n",
    "print(f\"最大池化后的特征图：\\n{max_pool_output}\")\n",
    "print(f\"\\n示例2(2x2平均池化核，填充为0，步长为2)：{avg_pool_output.shape}\")\n",
    "print(f\"平均池化后的特征图：\\n{avg_pool_output}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9c42e278-1c75-4f38-ac47-022b21177dbb",
   "metadata": {},
   "source": [
    "### 可视化卷积核和卷积操作"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9d6d9402-4a25-48ce-83ae-9ea7dad50572",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "\n",
    "# 加载MNIST数据集\n",
    "transform = transforms.ToTensor()\n",
    "mnist_dataset = torchvision.datasets.MNIST(root='./data', train=True, transform=transform, download=True)\n",
    "data_loader = torch.utils.data.DataLoader(dataset=mnist_dataset, batch_size=1, shuffle=True)\n",
    "\n",
    "# 获取一张示例图像\n",
    "image, label = next(iter(data_loader))  # 形状：(1, 1, 28, 28)\n",
    "\n",
    "# 定义卷积层：1输入通道，4输出通道，3x3核\n",
    "conv_layer = nn.Conv2d(in_channels=1, out_channels=4, kernel_size=3, stride=1, padding=1)\n",
    "\n",
    "# 前向传播\n",
    "with torch.no_grad():\n",
    "    feature_maps = conv_layer(image)  # 形状：(1, 4, 28, 28)\n",
    "\n",
    "# 获取卷积核权重\n",
    "weights = conv_layer.weight  # 形状：(4, 1, 3, 3)\n",
    "\n",
    "# 可视化卷积核权重和特征图\n",
    "plt.figure(figsize=(12, 6))\n",
    "\n",
    "# 可视化卷积核（4个3x3核）\n",
    "for i in range(4):\n",
    "    plt.subplot(2, 4, i + 1)\n",
    "    plt.title(f'Kernel {i+1}')\n",
    "    plt.imshow(weights[i, 0].detach().numpy(), cmap='gray')\n",
    "    plt.axis('off')\n",
    "\n",
    "# 可视化特征图（4张28x28特征图）\n",
    "for i in range(4):\n",
    "    plt.subplot(2, 4, i + 5)\n",
    "    plt.title(f'Feature Map {i+1}')\n",
    "    plt.imshow(feature_maps[0, i].detach().numpy(), cmap='gray')\n",
    "    plt.axis('off')\n",
    "\n",
    "# 显示输入图像\n",
    "plt.figure(figsize=(4, 4))\n",
    "plt.title('Input Image')\n",
    "plt.imshow(image[0, 0].detach().numpy(), cmap='gray')\n",
    "plt.axis('off')\n",
    "\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5903dbb-842a-460e-84bf-32ce188bca32",
   "metadata": {},
   "source": [
    "### 可视化卷积核和卷积操作示例\n",
    "\n",
    "**手动设置卷积核：**\n",
    "- 设置了4个特定卷积核：\n",
    "  - 核1：垂直边缘检测（Sobel算子）。\n",
    "  - 核2：水平边缘检测（Sobel算子）。\n",
    "  - 核3：拉普拉斯算子，增强边缘和细节。\n",
    "  - 核4：均值模糊，平滑图像。\n",
    "- 这些核使特征图更直观，易于理解卷积效果（如突出边缘或模糊区域）。\n",
    "\n",
    "**ReLU激活：**\n",
    "- 在卷积后应用nn.ReLU()，将负值置为0，增强特征图的非线性，改善灰度可视化效果（负值不再显示为黑色）。\n",
    "\n",
    "**特征图归一化：**\n",
    "- 对每个特征图进行归一化到[0, 1]，避免Matplotlib自动归一化导致的不一致显示\n",
    "\n",
    "**预期输出**\n",
    "\n",
    "- 第1个子图：原始MNIST图像（固定为第一张，可能是数字“5”），标题显示其标签。\n",
    "- 第2-5个子图：4个3×3卷积核的灰度图：\n",
    "  - 核1：垂直边缘检测，突出纵向边缘。\n",
    "  - 核2：水平边缘检测，突出横向边缘。\n",
    "  - 核3：拉普拉斯算子，突出细节和边缘。\n",
    "  - 核4：均值模糊，显示平滑效果。\n",
    "- 第6-10个子图：4个28×28特征图，对应各卷积核的输出：\n",
    "  - 特征图经过ReLU和归一化，白色表示高激活，黑色表示低激活（或零）。\n",
    "  - 垂直/水平边缘核会突出图像中的边缘，模糊核会平滑图像，拉普拉斯核会增强细节。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "150f91cd-42e4-4568-96b2-b237da6c13d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "\n",
    "# 加载MNIST数据集\n",
    "transform = transforms.ToTensor()  # 转换为张量并归一化到[0, 1]\n",
    "mnist_dataset = torchvision.datasets.MNIST(root='./data', train=True, transform=transform, download=True)\n",
    "\n",
    "# 固定获取第一张图像（索引0），确保结果可重复\n",
    "image, label = mnist_dataset[1]  # 形状：(1, 28, 28)\n",
    "image = image.unsqueeze(0)  # 增加batch维度，形状：(1, 1, 28, 28)\n",
    "\n",
    "# 定义卷积层：1输入通道，4输出通道，3x3核，步幅1，填充1\n",
    "conv_layer = nn.Conv2d(in_channels=1, out_channels=4, kernel_size=3, stride=1, padding=1)\n",
    "\n",
    "# 手动设置卷积核权重以提取特定特征\n",
    "with torch.no_grad():\n",
    "    conv_layer.weight[0] = torch.tensor([[[-1., 0., 1.], [-2., 0., 2.], [-1., 0., 1.]]])  # 垂直边缘\n",
    "    conv_layer.weight[1] = torch.tensor([[[1., 2., 1.], [0., 0., 0.], [-1., -2., -1.]]])  # 水平边缘\n",
    "    conv_layer.weight[2] = torch.tensor([[[0., 1., 0.], [1., -4., 1.], [0., 1., 0.]]])  # 拉普拉斯算子\n",
    "    conv_layer.weight[3] = torch.tensor([[[1., 1., 1.], [1., 1., 1.], [1., 1., 1.]]]) / 9.0  # 均值模糊\n",
    "    conv_layer.bias.fill_(0)  # 偏置设为0\n",
    "\n",
    "# 定义ReLU激活函数\n",
    "relu = nn.ReLU()\n",
    "\n",
    "# 前向传播\n",
    "with torch.no_grad():\n",
    "    feature_maps = relu(conv_layer(image))  # 形状：(1, 4, 28, 28)\n",
    "\n",
    "# 获取卷积核权重\n",
    "weights = conv_layer.weight  # 形状：(4, 1, 3, 3)\n",
    "\n",
    "# 可视化：统一2x5网格\n",
    "plt.figure(figsize=(15, 6))\n",
    "\n",
    "# 可视化输入图像\n",
    "plt.subplot(2, 5, 1)\n",
    "plt.title(f'Input Image (Label: {label})')\n",
    "plt.imshow(image[0, 0].detach().numpy(), cmap='gray')\n",
    "plt.axis('off')\n",
    "\n",
    "# 可视化卷积核（4个3x3核）\n",
    "for i in range(4):\n",
    "    plt.subplot(2, 5, i + 2)\n",
    "    plt.title(f'Kernel {i+1}')\n",
    "    plt.imshow(weights[i, 0].detach().numpy(), cmap='gray')\n",
    "    plt.axis('off')\n",
    "\n",
    "# 可视化特征图（4张28x28特征图）\n",
    "for i in range(4):\n",
    "    plt.subplot(2, 5, i + 7)\n",
    "    plt.title(f'Feature Map {i+1}')\n",
    "    feature_map = feature_maps[0, i].detach().numpy()\n",
    "    # 归一化特征图到[0, 1]\n",
    "    feature_map = (feature_map - feature_map.min()) / (feature_map.max() - feature_map.min() + 1e-8)\n",
    "    plt.imshow(feature_map, cmap='gray')\n",
    "    plt.axis('off')\n",
    "\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b8956f7b-2f48-4b28-9603-12e684dd5f7a",
   "metadata": {},
   "source": [
    "### 池化操作示例"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "16afcf4e-fc34-4cd6-84a5-b94aa1ef4ff1",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "\n",
    "# 1. 加载MNIST数据集\n",
    "transform = transforms.ToTensor()\n",
    "mnist_dataset = torchvision.datasets.MNIST(root='./data', train=True, download=True, transform=transform)\n",
    "data_loader = torch.utils.data.DataLoader(mnist_dataset, batch_size=1, shuffle=True)\n",
    "\n",
    "# 2. 获取一张图像\n",
    "image, label = next(iter(data_loader))\n",
    "image = image.squeeze().numpy()  # 形状从 [1, 1, 28, 28] 转为 [28, 28]\n",
    "\n",
    "# 3. 应用最大池化和平均池化\n",
    "image_tensor = torch.tensor(image).unsqueeze(0).unsqueeze(0)  # 形状转为 [1, 1, 28, 28]\n",
    "\n",
    "# 最大池化\n",
    "max_pool = torch.nn.MaxPool2d(kernel_size=2, stride=2)\n",
    "max_pooled_image = max_pool(image_tensor).squeeze().numpy()  # 形状为 [14, 14]\n",
    "\n",
    "# 平均池化\n",
    "avg_pool = torch.nn.AvgPool2d(kernel_size=2, stride=2)\n",
    "avg_pooled_image = avg_pool(image_tensor).squeeze().numpy()  # 形状为 [14, 14]\n",
    "\n",
    "# 4. 可视化\n",
    "fig, axes = plt.subplots(1, 3, figsize=(15, 5))\n",
    "\n",
    "# 原始图像\n",
    "axes[0].imshow(image, cmap='gray')\n",
    "axes[0].set_title(f'Original Image (Label: {label.item()})')\n",
    "axes[0].axis('off')\n",
    "\n",
    "# 最大池化后的图像\n",
    "axes[1].imshow(max_pooled_image, cmap='gray')\n",
    "axes[1].set_title('Max Pooled Image (2x2)')\n",
    "axes[1].axis('off')\n",
    "\n",
    "# 平均池化后的图像\n",
    "axes[2].imshow(avg_pooled_image, cmap='gray')\n",
    "axes[2].set_title('Average Pooled Image (2x2)')\n",
    "axes[2].axis('off')\n",
    "\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "617e895c-b905-4548-b292-2d3eb5fb8279",
   "metadata": {},
   "source": [
    "### 单层卷积与多层连续卷积对比\n",
    "\n",
    "用于对比：\n",
    "- 一卷一池（OneConv-OnePool）\n",
    "- 三卷一池（ThreeConv-OnePool）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3700841a-cc15-4834-b290-ce4eca460264",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 第一步：导入依赖库\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torchvision.transforms as T\n",
    "import torchvision.models as models\n",
    "from PIL import Image\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import requests\n",
    "from io import BytesIO\n",
    "\n",
    "# 第二步：读取一张在线图像\n",
    "#url = \"https://zh.wikipedia.org/zh-cn/%E7%8C%AB#/media/File:Cat_November_2010-1a.jpg\"\n",
    "#response = requests.get(url)\n",
    "#img = Image.open(BytesIO(response.content)).convert(\"RGB\")\n",
    "\n",
    "# 第二步：读取本地图像\n",
    "img = Image.open(\"./computational_graph.png\").convert(\"RGB\")\n",
    "\n",
    "transform = T.Compose([\n",
    "    T.Resize((64, 64)),\n",
    "    T.ToTensor(),\n",
    "])\n",
    "input_tensor = transform(img).unsqueeze(0)  # Shape: (1, 3, 64, 64)\n",
    "\n",
    "# 第三步：定义两个CNN模型结构\n",
    "# 一卷一池结构\n",
    "class OneConvOnePool(nn.Module):\n",
    "    def __init__(self):\n",
    "        super().__init__()\n",
    "        self.features = nn.Sequential(\n",
    "            nn.Conv2d(3, 16, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(2)\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        return self.features(x)\n",
    "\n",
    "# 三卷一池结构\n",
    "class ThreeConvOnePool(nn.Module):\n",
    "    def __init__(self):\n",
    "        super().__init__()\n",
    "        self.features = nn.Sequential(\n",
    "            nn.Conv2d(3, 16, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.Conv2d(16, 16, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.Conv2d(16, 16, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(2)\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        return self.features(x)\n",
    "\n",
    "# 第四步：生成特征图\n",
    "model_one = OneConvOnePool()\n",
    "model_three = ThreeConvOnePool()\n",
    "\n",
    "with torch.no_grad():\n",
    "    feat_one = model_one(input_tensor)\n",
    "    feat_three = model_three(input_tensor)\n",
    "\n",
    "# 第五步：绘图函数\n",
    "def plot_feature_maps(features, title, num_channels=6):\n",
    "    plt.figure(figsize=(12, 4))\n",
    "    for i in range(num_channels):\n",
    "        plt.subplot(1, num_channels, i + 1)\n",
    "        plt.imshow(features[0, i].cpu().numpy(), cmap='viridis')\n",
    "        plt.axis(\"off\")\n",
    "    plt.suptitle(title)\n",
    "    plt.show()\n",
    "\n",
    "# 第六步：可视化\n",
    "plot_feature_maps(feat_one, \"OneConvOnePool：前6个特征图\")\n",
    "plot_feature_maps(feat_three, \"ThreeConvOnePool：前6个特征图\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a1b945f-a299-4ad4-a6d4-db5b041ca529",
   "metadata": {},
   "source": [
    "### 展平层示例\n",
    "\n",
    "#### torch.flatten()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "66873b2d-9f77-434d-9b9e-a969b85b580d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "\n",
    "# 假设一个批次中包含 4 张图片，每张图片经过卷积和池化后，\n",
    "# 得到 64 个特征图，每个特征图大小是 7x7\n",
    "feature_map = torch.randn(4, 64, 7, 7)\n",
    "print(f\"原始特征图形状: {feature_map.shape}\")\n",
    "\n",
    "# 使用 torch.flatten()，从维度 1 开始展平，保留批次维度\n",
    "flattened_feature = torch.flatten(feature_map, start_dim=1)\n",
    "print(f\"展平后特征形状: {flattened_feature.shape}\")\n",
    "# 展平后的特征总数是 64 * 7 * 7 = 3136\n",
    "print(f\"每个样本的特征总数: {flattened_feature.shape[1]}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "70ecb15c-bc95-4927-83b9-cfbb24a43a25",
   "metadata": {},
   "source": [
    "#### nn.Flatten"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3137c623-76db-4239-8559-9b7ee2e1d3d6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 模拟卷积层输出的特征图\n",
    "feature_map = torch.randn(2, 128, 4, 4) # 2个批次，128个通道，4x4特征图\n",
    "print(f\"原始特征图形状: {feature_map.shape}\")\n",
    "\n",
    "# 创建一个 Flatten 模块实例\n",
    "flatten_layer = nn.Flatten()\n",
    "\n",
    "# 将特征图通过展平层\n",
    "flattened_feature_nn = flatten_layer(feature_map)\n",
    "print(f\"展平后特征形状: {flattened_feature_nn.shape}\")\n",
    "print(f\"每个样本的特征总数: {flattened_feature_nn.shape[1]}\")\n",
    "\n",
    "# 在一个简单的CNN模型中使用 nn.Flatten\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.conv_block = nn.Sequential(\n",
    "            nn.Conv2d(3, 16, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2)\n",
    "        )\n",
    "\n",
    "        # 使用 nn.Flatten 将卷积池化后的多维输出展平\n",
    "        self.flatten = nn.Flatten() \n",
    "\n",
    "        # 假设输入图片是 32x32，经过 conv_block 后，特征图尺寸变为 16x16，通道数为 16\n",
    "        # 展平后特征数 = 16 (通道数) * 16 (高) * 16 (宽) = 4096\n",
    "        self.fc_layer = nn.Linear(16 * 16 * 16, 10) # 假设输出 10 个类别\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.conv_block(x)\n",
    "        print(f\"卷积池化后的中间形状: {x.shape}\") \n",
    "        x = self.flatten(x) # 展平操作\n",
    "        print(f\"展平后的形状: {x.shape}\")\n",
    "        x = self.fc_layer(x)\n",
    "        return x\n",
    "\n",
    "model = SimpleCNN()\n",
    "dummy_input = torch.randn(1, 3, 32, 32) # 1个批次，3通道，32x32图片\n",
    "output = model(dummy_input)\n",
    "print(f\"最终输出形状: {output.shape}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c04cf0de-50de-43a1-a3f4-a6f0834157ed",
   "metadata": {},
   "source": [
    "### CNN正向传播和反向传播"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02bbe345-01a6-4874-8845-8cd2805f02d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.nn.functional as F\n",
    "\n",
    "# ========= 定义模型 =========\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        # 输入：(1, 28, 28)\n",
    "        self.conv1 = nn.Conv2d(in_channels=1, out_channels=8, kernel_size=3, padding=1)   # 输出：(8, 28, 28)\n",
    "        self.pool = nn.MaxPool2d(2, 2)                                                   # 输出：(8, 14, 14)\n",
    "        self.conv2 = nn.Conv2d(8, 16, 3, padding=1)                                       # 输出：(16, 14, 14)\n",
    "        # 再池化后输出：(16, 7, 7)\n",
    "        self.fc1 = nn.Linear(16 * 7 * 7, 64)\n",
    "        self.fc2 = nn.Linear(64, 10)  # 假设10分类任务\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.pool(F.relu(self.conv1(x)))   # Conv1 + ReLU + Pool\n",
    "        x = self.pool(F.relu(self.conv2(x)))   # Conv2 + ReLU + Pool\n",
    "        x = x.view(-1, 16 * 7 * 7)              # 展平\n",
    "        x = F.relu(self.fc1(x))                # 全连接层1\n",
    "        x = self.fc2(x)                        # 输出层（未使用 softmax）\n",
    "        return x\n",
    "\n",
    "# ========= 初始化模型、损失函数、输入 =========\n",
    "model = SimpleCNN()\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "\n",
    "# 伪造一个 batch 数据：batch_size=4，单通道 28x28 图像\n",
    "inputs = torch.randn(4, 1, 28, 28)\n",
    "labels = torch.tensor([1, 0, 3, 2])  # 伪造目标标签（0-9）\n",
    "\n",
    "# ========= 正向传播 =========\n",
    "outputs = model(inputs)\n",
    "loss = criterion(outputs, labels)\n",
    "\n",
    "# ========= 反向传播 =========\n",
    "loss.backward()\n",
    "\n",
    "# ========= 打印每层参数的梯度 =========\n",
    "print(\"=== 每层的权重梯度 ===\")\n",
    "for name, param in model.named_parameters():\n",
    "    if param.requires_grad:\n",
    "        print(f\"{name}: grad shape = {param.grad.shape}\")\n",
    "        print(param.grad)  # 可加 .norm() 或 .mean() 简化输出\n",
    "        print('-' * 50)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7895e233-b920-4c5e-8170-680a93323125",
   "metadata": {},
   "source": [
    "### 简单示例\n",
    "\n",
    "对比使用展平层(nn.Flatten)和全局平均池化层（nn.AdaptiveAvgPool2d((1, 1)）对特征图进行展平操作的不同的效果。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ed11bec0-43c5-4c99-b460-04ccc16d2ed3",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        \n",
    "        # 卷积和池化层用于特征提取\n",
    "        self.features = nn.Sequential(\n",
    "            nn.Conv2d(3, 32, kernel_size=3, padding=1),  # Input: [N, 3, 32, 32] -> Output: [N, 32, 32, 32]\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2),      # Output: [N, 32, 16, 16]\n",
    "            \n",
    "            nn.Conv2d(32, 64, kernel_size=3, padding=1), # Output: [N, 64, 16, 16]\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2)       # Output: [N, 64, 8, 8]\n",
    "        )\n",
    "        \n",
    "        # 展平层：将 [N, 64, 8, 8] 展平为 [N, 64 * 8 * 8]\n",
    "        # 64 * 8 * 8 = 4096\n",
    "        self.flatten = nn.Flatten()\n",
    "        \n",
    "        # 全连接层：接收展平后的特征，并映射到类别数量\n",
    "        # in_features: 4096 (来自展平层)\n",
    "        # out_features: num_classes (例如10个类别)\n",
    "        self.classifier = nn.Sequential(\n",
    "            nn.Linear(64 * 8 * 8, 128), # 第一个全连接层\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(0.5),            # 加入 Dropout 进行正则化，防止过拟合\n",
    "            nn.Linear(128, num_classes) # 第二个全连接层，输出类别分数\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        # 1. 卷积和池化提取特征\n",
    "        x = self.features(x)\n",
    "        print(f\"形状在展平前: {x.shape}\") # 例如: torch.Size([N, 64, 8, 8])\n",
    "        \n",
    "        # 2. 展平特征\n",
    "        x = self.flatten(x)\n",
    "        print(f\"形状在展平后: {x.shape}\") # 例如: torch.Size([N, 4096])\n",
    "        \n",
    "        # 3. 通过全连接层进行分类\n",
    "        x = self.classifier(x)\n",
    "        return x\n",
    "\n",
    "# 创建模型实例\n",
    "model = SimpleCNN(num_classes=10)\n",
    "\n",
    "# 模拟输入图像 (批次大小4, 3通道, 32x32像素)\n",
    "dummy_input = torch.randn(4, 3, 32, 32)\n",
    "\n",
    "# 前向传播\n",
    "output = model(dummy_input)\n",
    "print(f\"最终输出形状: {output.shape}\") # 期望: torch.Size([4, 10])\n",
    "\n",
    "# 查看模型参数总量 (nn.Linear层会贡献大量参数)\n",
    "total_params = sum(p.numel() for p in model.parameters() if p.requires_grad)\n",
    "print(f\"模型总参数量: {total_params}\")\n",
    "\n",
    "# 如果使用全局平均池化替代展平，参数量会大幅减少\n",
    "class SimpleCNN_GAP(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(SimpleCNN_GAP, self).__init__()\n",
    "        self.features = nn.Sequential(\n",
    "            nn.Conv2d(3, 32, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2),\n",
    "            \n",
    "            nn.Conv2d(32, 64, kernel_size=3, padding=1),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2)\n",
    "        )\n",
    "        \n",
    "        # 使用全局平均池化，将每个 8x8 的特征图平均为一个值\n",
    "        # 输出形状变为 [N, 64, 1, 1]\n",
    "        self.global_avg_pool = nn.AdaptiveAvgPool2d((1, 1))\n",
    "        \n",
    "        # 全连接层：in_features 变为 64 (通道数)，而不是 64*8*8\n",
    "        self.classifier = nn.Sequential(\n",
    "            nn.Linear(64, 128), # 参数量大幅减少\n",
    "            nn.ReLU(),\n",
    "            nn.Dropout(0.5),\n",
    "            nn.Linear(128, num_classes)\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.features(x)\n",
    "        print(f\"\\n形状在全局池化前: {x.shape}\") # 例如: torch.Size([N, 64, 8, 8])\n",
    "        \n",
    "        x = self.global_avg_pool(x)\n",
    "        print(f\"形状在全局池化后: {x.shape}\") # 例如: torch.Size([N, 64, 1, 1])\n",
    "        \n",
    "        # 全局池化后，张量形状通常是 [N, C, 1, 1]，需要再次展平或移除单维度\n",
    "        x = x.view(x.size(0), -1) # 或使用 x.squeeze(2).squeeze(2) 如果不确定维度\n",
    "        print(f\"最终输入全连接层形状: {x.shape}\") # 例如: torch.Size([N, 64])\n",
    "        \n",
    "        x = self.classifier(x)\n",
    "        return x\n",
    "\n",
    "model_gap = SimpleCNN_GAP(num_classes=10)\n",
    "output_gap = model_gap(dummy_input)\n",
    "print(f\"最终输出形状 (GAP): {output_gap.shape}\")\n",
    "total_params_gap = sum(p.numel() for p in model_gap.parameters() if p.requires_grad)\n",
    "print(f\"模型总参数量 (使用GAP): {total_params_gap}\") # 会发现参数量显著减少"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d276e818-537e-46e9-a53f-bd2dfd3f17a0",
   "metadata": {},
   "source": [
    "### LeNet-5示例\n",
    "\n",
    "LeNet-5是由Yann LeCun等人于1998年提出的经典卷积神经网络（CNN）架构，最初设计用于手写数字识别（如MNIST数据集）。它包含卷积层、池化层、展平层和全连接层，结构简单但奠定了现代CNN的基础。\n",
    "\n",
    "**架构介绍**\n",
    "\n",
    "LeNet-5 针对MNIST数据集（28×28灰度图像，10类数字）设计，结构如下：\n",
    "\n",
    "- C1：卷积层，6个5×5卷积核，输出 [batch, 6, 28, 28]。\n",
    "- S2：平均池化层，2×2窗口，步幅2，输出 [batch, 6, 14, 14]。\n",
    "- C3：卷积层，16个5×5卷积核，输出 [batch, 16, 10, 10]。\n",
    "- S4：平均池化层，2×2窗口，步幅2，输出 [batch, 16, 5, 5]。\n",
    "- C5：卷积层（或视为全连接层），120个5×5卷积核，输出 [batch, 120, 1, 1]。\n",
    "- F6：全连接层，120→84，输出 [batch, 84]。\n",
    "- Output：全连接层，84→10，输出 [batch, 10]（10类概率）。\n",
    "\n",
    "**注意：**\n",
    "\n",
    "- 原始LeNet-5使用Sigmoid/Tanh激活函数，现代实现通常用ReLU以加速收敛。\n",
    "- 原始C5层可视为全连接层（因输入5×5，卷积后为1×1），现代实现可能直接用展平层+全连接层。\n",
    "- 输出层通常接Softmax（结合交叉熵损失）。\n",
    "\n",
    "以下示例提供了基于PyTorch的LeNet-5模型代码，包括模型定义、MNIST数据集加载、训练和测试流程，并结合之前的上下文（如MNIST示例）进行说明。代码将训练模型并输出训练/测试准确率及可视化结果。\n",
    "\n",
    "**模型结构**\n",
    "\n",
    "模型结构说明：\n",
    "- C1 (conv1)：nn.Conv2d(1, 6, kernel_size=5, padding=2)，输入 [batch, 1, 28, 28]，输出 [batch, 6, 28, 28]（padding=2 保持尺寸）。\n",
    "- S2 (pool1)：nn.AvgPool2d(kernel_size=2, stride=2)，输出 [batch, 6, 14, 14]。\n",
    "- C3 (conv2)：nn.Conv2d(6, 16, kernel_size=5)，输出 [batch, 16, 10, 10]（无填充，尺寸减小）。\n",
    "- S4 (pool2)：nn.AvgPool2d(kernel_size=2, stride=2)，输出 [batch, 16, 5, 5]。\n",
    "- Flatten：nn.Flatten()，将 [batch, 16, 5, 5] 展平为 [batch, 16*5*5] = [batch, 400]。\n",
    "- F5 (fc1)：nn.Linear(400, 120)，输出 [batch, 120]。\n",
    "- F6 (fc2)：nn.Linear(120, 84)，输出 [batch, 84]。\n",
    "- Output (fc3)：nn.Linear(84, 10)，输出 [batch, 10]（logits）。\n",
    "\n",
    "**数据加载**\n",
    "\n",
    "MNIST数据集：\n",
    "- 训练集：60000张28×28灰度图像。\n",
    "- 测试集：10000张。\n",
    "- 使用 transforms.ToTensor()将图像归一化到 [0, 1]。\n",
    "- 批次大小：64，适合GPU/CPU训练。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2ee0235a-6eeb-4203-b96d-1946b61888dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "\n",
    "# 设置随机种子\n",
    "torch.manual_seed(42)\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "\n",
    "# 定义LeNet-5模型\n",
    "class LeNet5(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(LeNet5, self).__init__()\n",
    "        self.conv1 = nn.Conv2d(1, 6, kernel_size=5, stride=1, padding=2)  # C1: [batch, 1, 28, 28] -> [batch, 6, 28, 28]\n",
    "        self.relu = nn.ReLU()\n",
    "        self.pool1 = nn.AvgPool2d(kernel_size=2, stride=2)  # S2: [batch, 6, 28, 28] -> [batch, 6, 14, 14]\n",
    "        self.conv2 = nn.Conv2d(6, 16, kernel_size=5, stride=1)  # C3: [batch, 6, 14, 14] -> [batch, 16, 10, 10]\n",
    "        self.pool2 = nn.AvgPool2d(kernel_size=2, stride=2)  # S4: [batch, 16, 10, 10] -> [batch, 16, 5, 5]\n",
    "        self.flatten = nn.Flatten()  # Flatten: [batch, 16, 5, 5] -> [batch, 16*5*5]\n",
    "        self.fc1 = nn.Linear(16 * 5 * 5, 120)  # C5/F5: [batch, 400] -> [batch, 120]\n",
    "        self.fc2 = nn.Linear(120, 84)  # F6: [batch, 120] -> [batch, 84]\n",
    "        self.fc3 = nn.Linear(84, 10)  # Output: [batch, 84] -> [batch, 10]\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.relu(self.conv1(x))\n",
    "        x = self.pool1(x)\n",
    "        conv2_out = x  # 保存中间特征图用于可视化\n",
    "        x = self.relu(self.conv2(x))\n",
    "        x = self.pool2(x)\n",
    "        x = self.flatten(x)\n",
    "        x = self.relu(self.fc1(x))\n",
    "        x = self.relu(self.fc2(x))\n",
    "        x = self.fc3(x)  # 输出logits，交叉熵损失自带Softmax\n",
    "        return x, conv2_out\n",
    "\n",
    "# 加载MNIST数据集\n",
    "transform = transforms.ToTensor()\n",
    "train_dataset = torchvision.datasets.MNIST(root='./data', train=True, transform=transform, download=True)\n",
    "test_dataset = torchvision.datasets.MNIST(root='./data', train=False, transform=transform, download=True)\n",
    "train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=64, shuffle=True)\n",
    "test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=64, shuffle=False)\n",
    "\n",
    "# 初始化模型、损失函数和优化器\n",
    "model = LeNet5().to(device)\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 训练模型\n",
    "num_epochs = 5\n",
    "train_losses = []\n",
    "train_accuracies = []\n",
    "for epoch in range(num_epochs):\n",
    "    model.train()\n",
    "    running_loss = 0.0\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    for images, labels in train_loader:\n",
    "        images, labels = images.to(device), labels.to(device)\n",
    "        optimizer.zero_grad()\n",
    "        outputs, _ = model(images)\n",
    "        loss = criterion(outputs, labels)\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        running_loss += loss.item()\n",
    "        _, predicted = torch.max(outputs, 1)\n",
    "        total += labels.size(0)\n",
    "        correct += (predicted == labels).sum().item()\n",
    "    epoch_loss = running_loss / len(train_loader)\n",
    "    epoch_acc = 100 * correct / total\n",
    "    train_losses.append(epoch_loss)\n",
    "    train_accuracies.append(epoch_acc)\n",
    "    print(f\"Epoch [{epoch+1}/{num_epochs}], Loss: {epoch_loss:.4f}, Accuracy: {epoch_acc:.2f}%\")\n",
    "\n",
    "# 测试模型\n",
    "model.eval()\n",
    "correct = 0\n",
    "total = 0\n",
    "with torch.no_grad():\n",
    "    for images, labels in test_loader:\n",
    "        images, labels = images.to(device), labels.to(device)\n",
    "        outputs, _ = model(images)\n",
    "        _, predicted = torch.max(outputs, 1)\n",
    "        total += labels.size(0)\n",
    "        correct += (predicted == labels).sum().item()\n",
    "test_acc = 100 * correct / total\n",
    "print(f\"Test Accuracy: {test_acc:.2f}%\")\n",
    "\n",
    "# 可视化第一个卷积层的卷积核和第二个卷积层的特征图\n",
    "model.eval()\n",
    "with torch.no_grad():\n",
    "    image, label = train_dataset[0]\n",
    "    image = image.unsqueeze(0).to(device)\n",
    "    _, conv2_out = model(image)\n",
    "\n",
    "# 可视化\n",
    "plt.figure(figsize=(15, 6))\n",
    "# 输入图像\n",
    "plt.subplot(2, 4, 1)\n",
    "plt.title(f'Input Image (Label: {label})')\n",
    "plt.imshow(image[0, 0].cpu().numpy(), cmap='gray')\n",
    "plt.axis('off')\n",
    "\n",
    "# 第一个卷积层卷积核\n",
    "weights = model.conv1.weight.cpu().detach().numpy()  # [6, 1, 5, 5]\n",
    "for i in range(min(3, weights.shape[0])):  # 显示前3个卷积核\n",
    "    plt.subplot(2, 4, i + 2)\n",
    "    plt.title(f'Conv1 Kernel {i+1}')\n",
    "    plt.imshow(weights[i, 0], cmap='gray')\n",
    "    plt.axis('off')\n",
    "\n",
    "# 第二个卷积层特征图\n",
    "feature_maps = conv2_out[0].cpu().detach().numpy()  # [6, 14, 14]\n",
    "for i in range(min(4, feature_maps.shape[0])):  # 显示前4个特征图\n",
    "    plt.subplot(2, 4, i + 5)\n",
    "    plt.title(f'Conv2 Feature {i+1}')\n",
    "    feature_map = feature_maps[i]\n",
    "    feature_map = (feature_map - feature_map.min()) / (feature_map.max() - feature_map.min() + 1e-8)\n",
    "    plt.imshow(feature_map, cmap='gray')\n",
    "    plt.axis('off')\n",
    "\n",
    "plt.tight_layout()\n",
    "plt.show()\n",
    "\n",
    "# 可视化训练损失和准确率\n",
    "plt.figure(figsize=(10, 4))\n",
    "plt.subplot(1, 2, 1)\n",
    "plt.plot(train_losses, label='Train Loss')\n",
    "plt.xlabel('Epoch')\n",
    "plt.ylabel('Loss')\n",
    "plt.title('Training Loss')\n",
    "plt.legend()\n",
    "plt.subplot(1, 2, 2)\n",
    "plt.plot(train_accuracies, label='Train Accuracy')\n",
    "plt.xlabel('Epoch')\n",
    "plt.ylabel('Accuracy (%)')\n",
    "plt.title('Training Accuracy')\n",
    "plt.legend()\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fae4f6ed-7fba-46ed-a996-da9f5de5734a",
   "metadata": {},
   "source": [
    "## 词嵌入示例"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0c84f5c-c7e7-4423-9813-893c4529a49f",
   "metadata": {},
   "source": [
    "### 随机初始化词嵌入矩阵\n",
    "\n",
    "需要事先安装nltk、datasets、string和collections等包。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c2b0c038-eeb8-43d5-8fcb-4515e31130bd",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "from datasets import load_dataset\n",
    "from collections import Counter\n",
    "import nltk\n",
    "from nltk.tokenize import word_tokenize\n",
    "import string\n",
    "\n",
    "# 下载 NLTK 数据\n",
    "nltk.download('punkt')\n",
    "\n",
    "# 加载 IMDB 数据集\n",
    "dataset = load_dataset('imdb')\n",
    "\n",
    "# 文本预处理\n",
    "def preprocess_text(text):\n",
    "    text = text.lower()\n",
    "    text = text.translate(str.maketrans('', '', string.punctuation))\n",
    "    text = ' '.join(text.split())\n",
    "    return word_tokenize(text)\n",
    "\n",
    "# 构建词汇表\n",
    "def build_vocab(texts, min_freq=5):\n",
    "    all_tokens = [token for text in texts for token in preprocess_text(text)]\n",
    "    token_counts = Counter(all_tokens)\n",
    "    vocab = {'<PAD>': 0, '<UNK>': 1}\n",
    "    vocab.update({token: idx + 2 for idx, (token, count) in enumerate(token_counts.items()) if count >= min_freq})\n",
    "    return vocab\n",
    "\n",
    "# 生成词汇表\n",
    "train_texts = [item['text'] for item in dataset['train']]\n",
    "vocab = build_vocab(train_texts, min_freq=5)\n",
    "\n",
    "# 创建词嵌入矩阵（随机初始化）\n",
    "vocab_size = len(vocab)\n",
    "embedding_dim = 100\n",
    "embedding_matrix = torch.randn(vocab_size, embedding_dim)  # 随机初始化\n",
    "embedding_matrix[0] = torch.zeros(embedding_dim)  # <PAD> 设置为零向量\n",
    "\n",
    "# 加载到嵌入层\n",
    "embedding_layer = nn.Embedding.from_pretrained(embedding_matrix, freeze=False)  # 可训练\n",
    "\n",
    "# 示例输入\n",
    "input_ids = torch.tensor([2, 3, 4], dtype=torch.long)  # 对应 \"the\", \"quick\", \"fox\"\n",
    "embeddings = embedding_layer(input_ids)\n",
    "print(\"词嵌入矩阵形状:\", embedding_matrix.shape)  # [vocab_size, embedding_dim]\n",
    "print(\"嵌入向量形状:\", embeddings.shape)  # [seq_len, embedding_dim]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9648cc78-7605-4ba4-86d5-1f301576c2e8",
   "metadata": {},
   "source": [
    "### 使用预训练 GloVe 嵌入"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d11d8530-b97e-4373-b811-74cd6c219f14",
   "metadata": {},
   "outputs": [],
   "source": [
    "from torchtext.vocab import GloVe\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "from datasets import load_dataset\n",
    "from collections import Counter\n",
    "import nltk\n",
    "from nltk.tokenize import word_tokenize\n",
    "import string\n",
    "\n",
    "# 下载 NLTK 数据\n",
    "nltk.download('punkt')\n",
    "\n",
    "# 加载 IMDB 数据集\n",
    "dataset = load_dataset('imdb')\n",
    "\n",
    "# 文本预处理\n",
    "def preprocess_text(text):\n",
    "    text = text.lower()\n",
    "    text = text.translate(str.maketrans('', '', string.punctuation))\n",
    "    text = ' '.join(text.split())\n",
    "    return word_tokenize(text)\n",
    "\n",
    "# 构建词汇表\n",
    "def build_vocab(texts, min_freq=5):\n",
    "    all_tokens = [token for text in texts for token in preprocess_text(text)]\n",
    "    token_counts = Counter(all_tokens)\n",
    "    vocab = {'<PAD>': 0, '<UNK>': 1}\n",
    "    vocab.update({token: idx + 2 for idx, (token, count) in enumerate(token_counts.items()) if count >= min_freq})\n",
    "    return vocab\n",
    "\n",
    "# 生成词汇表\n",
    "train_texts = [item['text'] for item in dataset['train']]\n",
    "vocab = build_vocab(train_texts, min_freq=5)\n",
    "\n",
    "# 加载 GloVe 嵌入\n",
    "glove = GloVe(name='6B', dim=300)\n",
    "\n",
    "# 创建词嵌入矩阵\n",
    "vocab_size = len(vocab)\n",
    "embedding_dim = 300\n",
    "embedding_matrix = torch.zeros(vocab_size, embedding_dim)\n",
    "for word, idx in vocab.items():\n",
    "    if word in glove.stoi:\n",
    "        embedding_matrix[idx] = glove.vectors[glove.stoi[word]]\n",
    "    else:\n",
    "        embedding_matrix[idx] = torch.randn(embedding_dim)  # 未登录词随机初始化\n",
    "\n",
    "# 加载到嵌入层\n",
    "embedding_layer = nn.Embedding.from_pretrained(embedding_matrix, freeze=True)  # 冻结\n",
    "\n",
    "# 示例输入\n",
    "input_ids = torch.tensor([2, 3, 4], dtype=torch.long)  # 对应 \"the\", \"quick\", \"fox\"\n",
    "embeddings = embedding_layer(input_ids)\n",
    "print(\"词嵌入矩阵形状:\", embedding_matrix.shape)  # [vocab_size, embedding_dim]\n",
    "print(\"嵌入向量形状:\", embeddings.shape)  # [seq_len, embedding_dim]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1e71fda-cf20-4f68-abb1-540c03f4f949",
   "metadata": {},
   "source": [
    "### 使用BERT嵌入"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5eaab56f-107c-4061-b8e1-26ffa4b46625",
   "metadata": {},
   "outputs": [],
   "source": [
    "from transformers import BertTokenizer, BertModel\n",
    "import torch\n",
    "\n",
    "# 加载 BERT 分词器和模型\n",
    "tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')\n",
    "model = BertModel.from_pretrained('bert-base-uncased')\n",
    "\n",
    "# 示例文本\n",
    "text = \"The quick brown fox\"\n",
    "encoding = tokenizer(text, return_tensors='pt', padding=True, truncation=True, max_length=128)\n",
    "input_ids = encoding['input_ids']  # [1, seq_len]\n",
    "\n",
    "# 获取嵌入（BERT 内部包含词嵌入矩阵）\n",
    "with torch.no_grad():\n",
    "    outputs = model(input_ids)\n",
    "    embeddings = outputs.last_hidden_state  # [1, seq_len, 768]\n",
    "\n",
    "print(\"嵌入向量形状:\", embeddings.shape)  # [1, seq_len, 768]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9c358259-ed78-44a1-bf4e-707559f38e88",
   "metadata": {},
   "source": [
    "## RNN示例"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "32d804b4-b824-47ea-92f1-687c5bcb591f",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### RNN的前向传播\n",
    "\n",
    "**RNN前向传播原理**\n",
    "对于单层单向RNN，在时间步 $ t $，隐藏状态 $ h_t $ 的计算公式为：\n",
    "$$h_t = \\tanh(W_{ih} \\cdot x_t + b_{ih} + W_{hh} \\cdot h_{t-1} + b_{hh})$$\n",
    "\n",
    "- $ x_t $：当前时间步的输入，形状为 (batch_size, input_size)。\n",
    "- $ h_{t-1} $：上一时间步的隐藏状态，形状为 (batch_size, hidden_size)。\n",
    "- $ W_{ih} $：输入到隐藏层的权重矩阵，形状为 (hidden_size, input_size)。\n",
    "- $ W_{hh} $：隐藏层到隐藏层的权重矩阵，形状为 (hidden_size, hidden_size)。\n",
    "- $ b_{ih}, b_{hh} $：偏置向量，形状为 (hidden_size,)。\n",
    "- $ \\tanh $：激活函数。\n",
    "\n",
    "**前向传播需要：**\n",
    "\n",
    "- 对每个时间步 $ t $，根据输入 $ x_t $ 和上一隐藏状态 $ h_{t-1} $ 计算 $ h_t $。\n",
    "- 保存所有时间步的隐藏状态 $ h_t $ 作为输出序列。\n",
    "- 返回最终隐藏状态 $ h_n $ 和输出序列。\n",
    "\n",
    "以下是用PyTorch手动实现RNN前向传播的代码示例，假设输入序列长度为3，批量大小为2，输入维度为4，隐藏状态维度为5。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9d07ba90-ec58-4f3d-a66a-e01c60b72050",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 设置随机种子以确保结果可复现\n",
    "torch.manual_seed(42)\n",
    "\n",
    "# 参数设置\n",
    "batch_size = 2\n",
    "seq_len = 3\n",
    "input_size = 4\n",
    "hidden_size = 5\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)  # 形状: (2, 3, 4)\n",
    "h0 = torch.zeros(batch_size, hidden_size)            # 初始隐藏状态: (2, 5)\n",
    "\n",
    "# 权重和偏置（随机初始化，模拟RNN参数）\n",
    "W_ih = torch.randn(hidden_size, input_size)  # 形状: (5, 4)\n",
    "W_hh = torch.randn(hidden_size, hidden_size) # 形状: (5, 5)\n",
    "b_ih = torch.randn(hidden_size)             # 形状: (5,)\n",
    "b_hh = torch.randn(hidden_size)             # 形状: (5,)\n",
    "\n",
    "# 手动实现RNN前向传播\n",
    "def manual_rnn_forward(input, h0, W_ih, W_hh, b_ih, b_hh):\n",
    "    # input: (batch_size, seq_len, input_size)\n",
    "    # h0: (batch_size, hidden_size)\n",
    "    outputs = []  # 保存每个时间步的隐藏状态\n",
    "    h_t = h0      # 当前隐藏状态，初始化为 h0\n",
    "\n",
    "    # 遍历序列的每个时间步\n",
    "    for t in range(input.size(1)):\n",
    "        x_t = input[:, t, :]  # 当前时间步输入: (batch_size, input_size)\n",
    "        # 计算 h_t = tanh(W_ih * x_t + b_ih + W_hh * h_{t-1} + b_hh)\n",
    "        h_t = torch.tanh(\n",
    "            torch.matmul(x_t, W_ih.t()) + b_ih + torch.matmul(h_t, W_hh.t()) + b_hh\n",
    "        )\n",
    "        outputs.append(h_t.unsqueeze(1))  # 保存 h_t，形状: (batch_size, 1, hidden_size)\n",
    "\n",
    "    # 拼接所有时间步的输出\n",
    "    output = torch.cat(outputs, dim=1)  # 形状: (batch_size, seq_len, hidden_size)\n",
    "    h_n = h_t  # 最后一个时间步的隐藏状态\n",
    "\n",
    "    return output, h_n\n",
    "\n",
    "# 执行手动前向传播\n",
    "manual_output, manual_h_n = manual_rnn_forward(input, h0, W_ih, W_hh, b_ih, b_hh)\n",
    "\n",
    "# 输出形状\n",
    "print(\"手动RNN输出形状:\", manual_output.shape)  # (2, 3, 5)\n",
    "print(\"手动RNN最终隐藏状态形状:\", manual_h_n.shape)  # (2, 5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2195543d-440e-4013-90c1-7a5ebdf0b202",
   "metadata": {},
   "source": [
    "#### 用RNN的自动传播机制进行验证"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "437f4179-95ce-488a-80ff-7c19319c4012",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 定义PyTorch RNN层\n",
    "rnn = nn.RNN(input_size=input_size, hidden_size=hidden_size, batch_first=True, nonlinearity='tanh')\n",
    "\n",
    "# 设置相同的权重和偏置\n",
    "with torch.no_grad():\n",
    "    rnn.weight_ih_l0.copy_(W_ih)\n",
    "    rnn.weight_hh_l0.copy_(W_hh)\n",
    "    rnn.bias_ih_l0.copy_(b_ih)\n",
    "    rnn.bias_hh_l0.copy_(b_hh)\n",
    "\n",
    "# 执行PyTorch RNN前向传播\n",
    "pytorch_output, pytorch_h_n = rnn(input, h0.unsqueeze(0))  # h0需扩展为 (1, batch_size, hidden_size)\n",
    "\n",
    "# 输出形状\n",
    "print(\"PyTorch RNN输出形状:\", pytorch_output.shape)  # (2, 3, 5)\n",
    "print(\"PyTorch RNN最终隐藏状态形状:\", pytorch_h_n.shape)  # (1, 2, 5)\n",
    "\n",
    "# 验证手动实现与PyTorch结果一致\n",
    "print(\"输出是否一致:\", torch.allclose(manual_output, pytorch_output, atol=1e-6))\n",
    "print(\"最终隐藏状态是否一致:\", torch.allclose(manual_h_n, pytorch_h_n.squeeze(0), atol=1e-6))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fdb8e202-28bc-4fea-ae2c-5ca9751924cc",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### 基于NumPy的RNN示例\n",
    "\n",
    "带前向传播和反向传播过程。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6377e1e9-6182-423c-9389-b1380a448dfe",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "np.random.seed(0)\n",
    "\n",
    "# 超参数\n",
    "input_size = 1\n",
    "hidden_size = 10\n",
    "output_size = 1\n",
    "learning_rate = 0.01\n",
    "epochs = 100\n",
    "\n",
    "# 初始化权重\n",
    "Wxh = np.random.randn(hidden_size, input_size) * 0.1\n",
    "Whh = np.random.randn(hidden_size, hidden_size) * 0.1\n",
    "Why = np.random.randn(output_size, hidden_size) * 0.1\n",
    "bh = np.zeros((hidden_size, 1))\n",
    "by = np.zeros((output_size, 1))\n",
    "\n",
    "# 合成简单数据\n",
    "T = 10  # 序列长度\n",
    "x_seq = np.linspace(0, 2*np.pi, T).reshape(T, 1, 1)  # shape: [T, batch, input_size]\n",
    "y_seq = np.sin(x_seq)                               # shape: [T, batch, output_size]\n",
    "\n",
    "def tanh(x):\n",
    "    return np.tanh(x)\n",
    "\n",
    "def dtanh(x):\n",
    "    return 1 - np.tanh(x)**2\n",
    "\n",
    "# 训练循环\n",
    "for epoch in range(epochs):\n",
    "    # 初始化隐藏状态\n",
    "    h_prev = np.zeros((hidden_size, 1))\n",
    "    \n",
    "    xs, hs, ys, targets = {}, {}, {}, {}\n",
    "    hs[-1] = h_prev\n",
    "    loss = 0\n",
    "    \n",
    "    # ----------- 前向传播 -------------\n",
    "    for t in range(T):\n",
    "        x_t = x_seq[t]  # shape: (1, 1)\n",
    "        y_target = y_seq[t]\n",
    "\n",
    "        xs[t] = x_t\n",
    "        hs[t] = tanh(np.dot(Wxh, x_t) + np.dot(Whh, hs[t-1]) + bh)\n",
    "        ys[t] = np.dot(Why, hs[t]) + by\n",
    "        targets[t] = y_target\n",
    "\n",
    "        loss += 0.5 * np.sum((ys[t] - y_target) ** 2)\n",
    "\n",
    "    # ----------- 反向传播 (BPTT) -------------\n",
    "    dWxh = np.zeros_like(Wxh)\n",
    "    dWhh = np.zeros_like(Whh)\n",
    "    dWhy = np.zeros_like(Why)\n",
    "    dbh = np.zeros_like(bh)\n",
    "    dby = np.zeros_like(by)\n",
    "    dh_next = np.zeros_like(hs[0])\n",
    "\n",
    "    for t in reversed(range(T)):\n",
    "        dy = ys[t] - targets[t]  # dL/dy\n",
    "        dWhy += np.dot(dy, hs[t].T)\n",
    "        dby += dy\n",
    "\n",
    "        dh = np.dot(Why.T, dy) + dh_next  # 误差传回隐藏层\n",
    "        dh_raw = dtanh(np.dot(Wxh, xs[t]) + np.dot(Whh, hs[t-1]) + bh) * dh  # dL/dh_t * tanh'\n",
    "        dbh += dh_raw\n",
    "        dWxh += np.dot(dh_raw, xs[t].T)\n",
    "        dWhh += np.dot(dh_raw, hs[t-1].T)\n",
    "        dh_next = np.dot(Whh.T, dh_raw)\n",
    "\n",
    "    # ----------- 梯度裁剪 (可选) -------------\n",
    "    for dparam in [dWxh, dWhh, dWhy, dbh, dby]:\n",
    "        np.clip(dparam, -1, 1, out=dparam)\n",
    "\n",
    "    # ----------- 参数更新 -------------\n",
    "    Wxh -= learning_rate * dWxh\n",
    "    Whh -= learning_rate * dWhh\n",
    "    Why -= learning_rate * dWhy\n",
    "    bh -= learning_rate * dbh\n",
    "    by -= learning_rate * dby\n",
    "\n",
    "    # ----------- 输出损失 -------------\n",
    "    if epoch % 10 == 0 or epoch == epochs - 1:\n",
    "        print(f\"Epoch {epoch}: Loss = {loss:.4f}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88011617-7b16-4cec-82cb-98da07e9a956",
   "metadata": {},
   "source": [
    "### 简单RNN示例\n",
    "\n",
    "#### 单层单向RNN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c58d5a78-677b-442a-b7f1-866f836ee7f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 参数设置\n",
    "input_size = 10    # 输入特征维度\n",
    "hidden_size = 20   # 隐藏状态维度\n",
    "batch_size = 32    # 批量大小\n",
    "seq_len = 50       # 序列长度\n",
    "\n",
    "# 定义RNN层\n",
    "rnn = nn.RNN(input_size=input_size, hidden_size=hidden_size, batch_first=True, nonlinearity='tanh')\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)  # (batch_size, seq_len, input_size)\n",
    "h0 = torch.randn(1, batch_size, hidden_size)          # 初始隐藏状态 (num_layers, batch_size, hidden_size)\n",
    "\n",
    "# 前向传播\n",
    "output, h_n = rnn(input, h0)\n",
    "\n",
    "# 输出形状\n",
    "print(output.shape)  # (batch_size, seq_len, hidden_size) = (32, 50, 20)\n",
    "print(h_n.shape)    # (num_layers, batch_size, hidden_size) = (1, 32, 20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c84a5e71-c2dc-484d-aa5e-129219c9486f",
   "metadata": {},
   "source": [
    "#### 多层双向RNN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a163eb32-cc11-4877-9579-acb8c8081a5c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 参数设置\n",
    "input_size = 10\n",
    "hidden_size = 20\n",
    "num_layers = 2\n",
    "batch_size = 32\n",
    "seq_len = 50\n",
    "\n",
    "# 定义双向RNN\n",
    "rnn = nn.RNN(input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, \n",
    "             batch_first=True, bidirectional=True, dropout=0.3, nonlinearity='relu')\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)\n",
    "h0 = torch.randn(2 * 2, batch_size, hidden_size)  # 2层 * 2方向\n",
    "\n",
    "# 前向传播\n",
    "output, h_n = rnn(input, h0)\n",
    "\n",
    "# 输出形状\n",
    "print(output.shape)  # (batch_size, seq_len, num_directions * hidden_size) = (32, 50, 40)\n",
    "print(h_n.shape)    # (num_layers * num_directions, batch_size, hidden_size) = (4, 32, 20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b9b0f153-28d6-4c73-bd05-7dd44ab1779b",
   "metadata": {},
   "source": [
    "#### 处理变长输入序列"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "93427a81-c4be-4420-8768-87216cb35f55",
   "metadata": {},
   "outputs": [],
   "source": [
    "from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence\n",
    "\n",
    "# 模拟变长序列\n",
    "input = torch.randn(batch_size, seq_len, input_size)\n",
    "lengths = torch.randint(10, seq_len + 1, (batch_size,))  # 每个序列的实际长度\n",
    "\n",
    "# 按长度降序排序\n",
    "lengths, indices = lengths.sort(descending=True)\n",
    "input = input[indices]\n",
    "\n",
    "# 定义RNN\n",
    "rnn = nn.RNN(input_size=input_size, hidden_size=hidden_size, batch_first=True)\n",
    "\n",
    "# 打包序列\n",
    "packed_input = pack_padded_sequence(input, lengths, batch_first=True, enforce_sorted=True)\n",
    "\n",
    "# 前向传播\n",
    "output, h_n = rnn(packed_input)\n",
    "\n",
    "# 解包输出\n",
    "output, output_lengths = pad_packed_sequence(output, batch_first=True)\n",
    "\n",
    "print(output.shape)  # (batch_size, seq_len, hidden_size)\n",
    "print(h_n.shape)    # (1, batch_size, hidden_size)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1133fbd-10e7-491c-8045-a25b8b6c41ba",
   "metadata": {},
   "source": [
    "#### RNN的权重访问\n",
    "\n",
    "- RNN层的权重矩阵可以通过 rnn.weight_ih_l[k]（输入到隐藏层的权重）和 rnn.weight_hh_l[k]（隐藏层到隐藏层的权重）访问，其中 k 是层索引。\n",
    "- 偏置项通过 rnn.bias_ih_l[k] 和 rnn.bias_hh_l[k] 访问。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "754a43eb-c3e8-4ef9-ab9f-bb19accaa678",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 访问第一层RNN的权重\n",
    "print(rnn.weight_ih_l0.shape)  # (hidden_size, input_size)\n",
    "print(rnn.weight_hh_l0.shape)  # (hidden_size, hidden_size)\n",
    "print(rnn.bias_ih_l0.shape)    # (hidden_size,)\n",
    "print(rnn.bias_hh_l0.shape)    # (hidden_size,)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a23d487f-f3c5-4bb4-af40-cc4136dec2e7",
   "metadata": {},
   "source": [
    "#### 带前向传播和反向传播的示例\n",
    "\n",
    "单次epoch。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "97983531-3fdb-4e35-82d0-f05fea02a627",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "\n",
    "# 定义模型\n",
    "class RNNClassifier(nn.Module):\n",
    "    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, num_layers=1):\n",
    "        super(RNNClassifier, self).__init__()\n",
    "        # Embedding层\n",
    "        self.embedding = nn.Embedding(num_embeddings=vocab_size, embedding_dim=embedding_dim, padding_idx=0)\n",
    "        # LSTM层\n",
    "        self.rnn = nn.LSTM(input_size=embedding_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True)\n",
    "        # 全连接层\n",
    "        self.fc = nn.Linear(hidden_dim, output_dim)\n",
    "    \n",
    "    def forward(self, text):\n",
    "        # text: [batch_size, seq_len]\n",
    "        embedded = self.embedding(text)  # [batch_size, seq_len, embedding_dim]\n",
    "        output, (h_n, c_n) = self.rnn(embedded)  # output: [batch_size, seq_len, hidden_dim]\n",
    "        # 取最后一个时间步的隐藏状态\n",
    "        final_hidden = h_n[-1, :, :]  # [batch_size, hidden_dim]\n",
    "        out = self.fc(final_hidden)  # [batch_size, output_dim]\n",
    "        return out\n",
    "\n",
    "# 参数设置\n",
    "vocab_size = 1000  # 词汇表大小\n",
    "embedding_dim = 100  # 嵌入向量维度\n",
    "hidden_dim = 128  # 隐藏状态维度\n",
    "output_dim = 2  # 分类数（例如二分类）\n",
    "num_layers = 2  # RNN层数\n",
    "batch_size = 32\n",
    "seq_len = 10\n",
    "\n",
    "# 实例化模型\n",
    "model = RNNClassifier(vocab_size, embedding_dim, hidden_dim, output_dim, num_layers)\n",
    "\n",
    "# 模拟输入数据\n",
    "input_text = torch.randint(0, vocab_size, (batch_size, seq_len))  # 随机索引 [batch_size, seq_len]\n",
    "labels = torch.randint(0, output_dim, (batch_size,))  # 随机标签 [batch_size]\n",
    "\n",
    "# 定义损失函数和优化器\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 训练一步\n",
    "model.train()\n",
    "optimizer.zero_grad()\n",
    "output = model(input_text)\n",
    "loss = criterion(output, labels)\n",
    "loss.backward()\n",
    "optimizer.step()\n",
    "\n",
    "print(\"Input Shape:\", input_text.shape)  # torch.Size([32, 10])\n",
    "print(\"Output Shape:\", output.shape)      # torch.Size([32, 2])\n",
    "print(\"Loss:\", loss.item())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "50c5fe27-a1cc-4b06-a142-ebe44a00f011",
   "metadata": {},
   "source": [
    "### 基于IMDB数据集的RNN模型实现示例\n",
    "\n",
    "我们将使用PyTorch实现一个简单的RNN模型，完成IMDB电影评论的情感分析任务（二分类：正面或负面）。下面的代码包括数据加载、预处理、模型定义、训练和评估等功能的定义。\n",
    "\n",
    "**提示**： 评论长度可变，需处理词汇表和填充。\n",
    "\n",
    "1. 数据预处理：\n",
    "\n",
    "- 数据集：使用Keras的imdb.load_data加载IMDB数据集，限制词汇表大小为10,000。\n",
    "- 序列处理：将评论截断或填充到固定长度（max_len=500），确保输入形状一致。\n",
    "- 自定义数据集：IMDBDataset类将数据转换为PyTorch张量，方便DataLoader使用。\n",
    "- DataLoader：批量加载数据，batch_size=32，打乱训练数据以提高泛化能力。\n",
    "\n",
    "2. 模型结构：\n",
    "\n",
    "- 嵌入层（nn.Embedding）：将单词索引转换为128维词嵌入向量。\n",
    "- RNN层（nn.RNN）：单层RNN，隐藏状态维度为256，使用tanh激活函数，batch_first=True。\n",
    "- 全连接层（nn.Linear）：将最后一个时间步的隐藏状态映射到1维输出。\n",
    "- Sigmoid：将输出压缩到[0, 1]，表示正面评论的概率。\n",
    "\n",
    "\n",
    "3. 训练过程：\n",
    "\n",
    "- 损失函数：BCELoss（二分类交叉熵），适合二分类任务。\n",
    "- 优化器：Adam，学习率为0.001。\n",
    "- 梯度裁剪：通过clip_grad_norm_限制梯度最大范数为1，防止梯度爆炸。\n",
    "- 训练5个epoch：每次迭代计算损失、反向传播、更新参数。\n",
    "\n",
    "\n",
    "4. 评估：\n",
    "\n",
    "- 计算测试集准确率，预测值大于0.5为正面，否则为负面。\n",
    "- 输出测试集准确率，评估模型性能。\n",
    "\n",
    "**注意事项**\n",
    "\n",
    "1. 词典限制：\n",
    "\n",
    "- IMDB词典只包含训练数据中的单词，未见词（OOV）映射到索引0，可能影响预测准确性。\n",
    "- 测试评论应尽量使用常见英语词汇，避免生僻词。\n",
    "\n",
    "2. 序列长度：\n",
    "\n",
    "- 测试评论被填充或截断到 max_len=500，用于确保同训练数据一致。\n",
    "- 短评论可能填充较多0，长评论可能丢失信息。\n",
    "\n",
    "\n",
    "3. 模型局限：\n",
    "\n",
    "- 标准RNN可能因梯度消失问题对长序列表现不佳，建议替换为 nn.LSTM 或 nn.GRU。\n",
    "- 增加num_layers或dropout可提高性能，但需更多计算资源。\n",
    "\n",
    "**扩展建议：**\n",
    "\n",
    "- 使用预训练词嵌入（如GloVe）替换 nn.Embedding。\n",
    "- 添加注意力机制或双向RNN（bidirectional=True）。\n",
    "- 增加验证集监控过拟合。\n",
    "\n",
    "\n",
    "提示：需要事先安装依赖的模块。重新安装，以确保几个模块间版本上的兼容：\n",
    "```bash\n",
    "pip install --force-reinstall numpy==2.1.3 tensorflow gensim scipy\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cad94f14-49e4-489c-807f-54560891aa69",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from tensorflow.keras.datasets import imdb\n",
    "import numpy as np\n",
    "import re\n",
    "import string\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "np.random.seed(42)\n",
    "\n",
    "# 1. 数据集准备\n",
    "class IMDBDataset(Dataset):\n",
    "    def __init__(self, data, labels, max_len=500):\n",
    "        self.data = data\n",
    "        self.labels = labels\n",
    "        self.max_len = max_len\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.data)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        # 截断或填充序列到 max_len\n",
    "        seq = self.data[idx][:self.max_len]\n",
    "        seq = np.pad(seq, (0, self.max_len - len(seq)), mode='constant') if len(seq) < self.max_len else seq\n",
    "        return torch.tensor(seq, dtype=torch.long), torch.tensor(self.labels[idx], dtype=torch.float)\n",
    "\n",
    "# 加载IMDB数据集和词典\n",
    "vocab_size = 20000  # 限制词汇表大小\n",
    "max_len = 500       # 最大序列长度\n",
    "(x_train, y_train), (x_test, y_test) = imdb.load_data(num_words=vocab_size)\n",
    "word_index = imdb.get_word_index()  # 获取单词到索引的映射\n",
    "\n",
    "# 创建PyTorch数据集\n",
    "train_dataset = IMDBDataset(x_train, y_train, max_len)\n",
    "test_dataset = IMDBDataset(x_test, y_test, max_len)\n",
    "\n",
    "# 创建数据加载器\n",
    "batch_size = 32\n",
    "train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)\n",
    "test_loader = DataLoader(test_dataset, batch_size=batch_size)\n",
    "\n",
    "# 2. 定义RNN模型\n",
    "class IMDBRNN(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size, output_size, num_layers=1):\n",
    "        super(IMDBRNN, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)  # 词嵌入层\n",
    "        self.rnn = nn.RNN(\n",
    "            input_size=embed_size,\n",
    "            hidden_size=hidden_size,\n",
    "            num_layers=num_layers,\n",
    "            batch_first=True,\n",
    "            nonlinearity='tanh'\n",
    "        )\n",
    "        self.fc = nn.Linear(hidden_size, output_size)  # 输出层\n",
    "        self.sigmoid = nn.Sigmoid()  # 用于二分类\n",
    "\n",
    "    def forward(self, x, h_0=None):\n",
    "        # x: [batch_size, seq_len]\n",
    "        x = self.embedding(x)  # [batch_size, seq_len, embed_size]\n",
    "        output, h_n = self.rnn(x, h_0)  # output: [batch_size, seq_len, hidden_size]\n",
    "        out = self.fc(output[:, -1, :])  # 取最后一个时间步: [batch_size, output_size]\n",
    "        out = self.sigmoid(out)  # [batch_size, 1]\n",
    "        return out\n",
    "\n",
    "# 3. 模型参数\n",
    "embed_size = 128    # 词嵌入维度\n",
    "hidden_size = 256   # 隐藏状态维度\n",
    "output_size = 1     # 输出维度（二分类）\n",
    "num_layers = 1      # RNN层数\n",
    "\n",
    "# 初始化模型、损失函数和优化器\n",
    "model = IMDBRNN(vocab_size, embed_size, hidden_size, output_size, num_layers)\n",
    "criterion = nn.BCELoss()  # 二分类交叉熵损失\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 4. 训练模型\n",
    "def train_model(model, train_loader, criterion, optimizer, epochs=5, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for data, labels in train_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(data)\n",
    "            loss = criterion(output, labels)\n",
    "            loss.backward()\n",
    "            # 梯度裁剪防止爆炸\n",
    "            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1)\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f\"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}\")\n",
    "\n",
    "# 5. 评估模型\n",
    "def evaluate_model(model, test_loader, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for data, labels in test_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            output = model(data)\n",
    "            predictions = (output >= 0.5).float()\n",
    "            correct += (predictions == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    accuracy = correct / total * 100\n",
    "    print(f\"Test Accuracy: {accuracy:.2f}%\")\n",
    "\n",
    "# 6. 测试功能：处理示例评论\n",
    "def text_to_sequence(text, word_index, max_len=500):\n",
    "    # 文本预处理：小写、去除标点\n",
    "    text = text.lower()\n",
    "    text = re.sub(f'[{string.punctuation}]', '', text)\n",
    "    # 分词\n",
    "    tokens = text.split()\n",
    "    # 转换为单词索引\n",
    "    sequence = [word_index.get(word, 0) for word in tokens if word_index.get(word, 0) < vocab_size]\n",
    "    # 截断或填充\n",
    "    sequence = sequence[:max_len]\n",
    "    sequence = np.pad(sequence, (0, max_len - len(sequence)), mode='constant') if len(sequence) < max_len else sequence\n",
    "    return torch.tensor([sequence], dtype=torch.long)  # [1, max_len]\n",
    "\n",
    "def test_model(model, texts, word_index, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        for text in texts:\n",
    "            # 转换为序列\n",
    "            seq = text_to_sequence(text, word_index, max_len)\n",
    "            seq = seq.to(device)\n",
    "            # 预测\n",
    "            output = model(seq)  # [1, 1]\n",
    "            prob = output.item()\n",
    "            sentiment = \"Positive\" if prob >= 0.5 else \"Negative\"\n",
    "            print(f\"Text: {text}\")\n",
    "            print(f\"Sentiment: {sentiment}, Probability: {prob:.4f}\\n\")\n",
    "\n",
    "# 7. 运行训练、评估和测试\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "# 训练轮数会影响模型的性能，例如对于本示例，5轮训练，对于下面测试示例中的第一个可能会得到错误结果，\n",
    "# 但10轮训练就可能会得到正确结果。\n",
    "train_model(model, train_loader, criterion, optimizer, epochs=30, device=device)\n",
    "evaluate_model(model, test_loader, device=device)\n",
    "\n",
    "# 示例评论\n",
    "test_texts = [\n",
    "    \"I absolutely love this movie, it's fantastic and thrilling!\",\n",
    "    \"This film was boring and a complete waste of time.\",\n",
    "    \"The plot was okay, but the acting was amazing.\",\n",
    "    \"Terrible movie, I hated every minute of it.\"\n",
    "]\n",
    "test_model(model, test_texts, word_index, device=device)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c80f5f7b-2aac-4d91-b960-1922459c26e0",
   "metadata": {},
   "source": [
    "#### 示例结果说明\n",
    "\n",
    "测试时使用的各示例可能会被错误地分类到不正确的类别中，这通常有几个可能的原因。\n",
    "\n",
    "1. 词汇表限制（Out-of-Vocabulary, OOV）：示例代码使用了imdb.load_data(num_words=20000)，这意味着它只加载了数据集中最常见的20000个单词。如果“fantastic”或“thrilling”这类词汇不在这个有限的词汇表中，它们就会被替换为未知的词汇标记（通常是0），在模型中得不到有效的学习。这会导致模型无法理解这些表达强烈正面情感的词汇，从而无法正确判断句子的情感倾向。\n",
    "\n",
    "2. 模型训练不足或欠拟合：示例代码中，模型只训练了10个周期（epochs）。对于一个大型数据集和复杂的任务来说，这可能不足以让模型充分学习到所有的情感模式。如果模型没有在训练过程中见到足够多的像“fantastic”或“thrilling”这样描述积极情感的词汇及其上下文，它就可能无法正确地将这些词与正面情感关联起来。\n",
    "\n",
    "3. 数据预处理或编码问题：虽然示例代码中的数据预处理看起来是标准的，但仍有可能存在细微的问题。例如，如果使用了自定义的词汇表或编码方式，并且在测试时没有保持一致，就可能导致模型无法识别这些词。在示例中，text_to_sequence函数将未在词典中的词汇映射为0，这与IMDB数据集本身的约定是一致的，但如果词典和训练数据在其他方面存在不匹配，也可能导致问题。\n",
    "\n",
    "4. 训练数据中的特殊情况或噪声：在IMDB数据集中，某些词汇可能在不同的上下文中被赋予了相反的情感。例如，在讽刺的评论中，“fantastic”可能会被用于表达负面情感。如果训练数据中存在大量的这类“噪声”，模型可能会产生错误的学习偏差。当然，这在IMDB数据集中并不常见，但也是一个理论上的可能性。\n",
    "\n",
    "为了解决以上这些类问题，可以尝试：\n",
    "- 增加训练周期：将epochs的值从10增加到20或30，让模型有更多时间学习。\n",
    "- 检查词汇表：确保所有重要的情感词汇都包含在你的词汇表中。你可以通过打印word_index来验证。\n",
    "- 调整模型参数：增加hidden_size或num_layers，让模型有更强的表达能力，以便更好地理解复杂的语言模式。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3598cbc1-c54a-4ae1-b708-62c269ad2575",
   "metadata": {},
   "source": [
    "## LSTM示例"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d26aa94e-60a3-4459-b4d7-135ac9bd00b7",
   "metadata": {},
   "source": [
    "### LSTM层的语法格式\n",
    "\n",
    "```python\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "lstm = nn.LSTM(input_size, hidden_size, num_layers=1, bias=True, batch_first=False, dropout=0, bidirectional=False)\n",
    "```\n",
    "\n",
    "**LSTM计算公式**\n",
    "\n",
    "对于单层单向LSTM，每个时间步的计算包括以下步骤：\n",
    "\n",
    "输入门：\n",
    "$$i_t = \\sigma(W_{ii} \\cdot x_t + b_{ii} + W_{hi} \\cdot h_{t-1} + b_{hi})$$\n",
    "\n",
    "遗忘门：\n",
    "$$f_t = \\sigma(W_{if} \\cdot x_t + b_{if} + W_{hf} \\cdot h_{t-1} + b_{hf})$$\n",
    "\n",
    "输出门：\n",
    "$$o_t = \\sigma(W_{io} \\cdot x_t + b_{io} + W_{ho} \\cdot h_{t-1} + b_{ho})$$\n",
    "\n",
    "候选记忆单元：\n",
    "$$\\tilde{c}_t = \\tanh(W_{ic} \\cdot x_t + b_{ic} + W_{hc} \\cdot h_{t-1} + b_{hc})$$\n",
    "\n",
    "记忆单元更新：\n",
    "$$c_t = f_t \\odot c_{t-1} + i_t \\odot \\tilde{c}_t$$\n",
    "\n",
    "隐藏状态更新：\n",
    "$$h_t = o_t \\odot \\tanh(c_t)$$\n",
    "\n",
    "- $ W_{ii}, W_{if}, W_{io}, W_{ic} $：输入到门的权重矩阵，形状为 (hidden_size, input_size)。\n",
    "- $ W_{hi}, W_{hf}, W_{ho}, W_{hc} $：隐藏层到门的权重矩阵，形状为 (hidden_size, hidden_size)。\n",
    "- $ b_{ii}, b_{if}, b_{io}, b_{ic}, b_{hi}, b_{hf}, b_{ho}, b_{hc} $：偏置向量，形状为 (hidden_size,)。\n",
    "- $ \\sigma $：sigmoid激活函数；$ \\odot $：逐元素乘法。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "afbfbf94-1759-4c7c-84c3-aaab688714a0",
   "metadata": {},
   "source": [
    "### 门控计算过程示例\n",
    "\n",
    "模拟1个时间步。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b2e2e61f-e0d9-4794-8b09-62524e1da049",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn.functional as F\n",
    "\n",
    "# 超参数设置\n",
    "input_size = 4     # 每个时间步输入维度\n",
    "hidden_size = 3    # 隐藏状态维度\n",
    "output_size = 2    # 最终输出维度（例如分类）\n",
    "\n",
    "# 模拟输入 x_t、前一时刻的 h_{t-1}, c_{t-1}\n",
    "x_t = torch.randn(input_size)       # (4,)\n",
    "h_prev = torch.randn(hidden_size)   # (3,)\n",
    "c_prev = torch.randn(hidden_size)   # (3,)\n",
    "\n",
    "# 初始化参数\n",
    "def init_param():\n",
    "    return torch.randn(hidden_size, input_size), torch.randn(hidden_size, hidden_size), torch.randn(hidden_size)\n",
    "\n",
    "# 门参数\n",
    "W_f, U_f, b_f = init_param()  # 忘记门\n",
    "W_i, U_i, b_i = init_param()  # 输入门\n",
    "W_o, U_o, b_o = init_param()  # 输出门\n",
    "W_c, U_c, b_c = init_param()  # 候选记忆\n",
    "\n",
    "# 输出层参数\n",
    "W_out = torch.randn(output_size, hidden_size)\n",
    "b_out = torch.randn(output_size)\n",
    "\n",
    "# ========== 各门计算 ==========\n",
    "f_t = torch.sigmoid(W_f @ x_t + U_f @ h_prev + b_f)  # 忘记门\n",
    "i_t = torch.sigmoid(W_i @ x_t + U_i @ h_prev + b_i)  # 输入门\n",
    "o_t = torch.sigmoid(W_o @ x_t + U_o @ h_prev + b_o)  # 输出门\n",
    "c_tilde = torch.tanh(W_c @ x_t + U_c @ h_prev + b_c) # 候选记忆\n",
    "\n",
    "# ========== 状态更新 ==========\n",
    "c_t = f_t * c_prev + i_t * c_tilde      # 当前单元状态\n",
    "h_t = o_t * torch.tanh(c_t)             # 当前隐藏状态\n",
    "\n",
    "# ========== 输出计算 ==========\n",
    "y_t = W_out @ h_t + b_out\n",
    "\n",
    "# 打印结果\n",
    "print(\"忘记门 f_t:\", f_t)\n",
    "print(\"输入门 i_t:\", i_t)\n",
    "print(\"候选记忆 c~_t:\", c_tilde)\n",
    "print(\"单元状态 c_t:\", c_t)\n",
    "print(\"输出门 o_t:\", o_t)\n",
    "print(\"隐藏状态 h_t:\", h_t)\n",
    "print(\"最终输出 y_t:\", y_t)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eeb8e603-2667-4b6c-b3e2-3220b12e57dc",
   "metadata": {},
   "source": [
    "### 多时间步+batch_size=3的LSTM手动前向传播实现\n",
    "\n",
    "**我们将模拟出如下环境：**\n",
    "- 输入序列：长度为 3\n",
    "- 每个序列样本：输入维度为 4\n",
    "- 批大小：3\n",
    "- 隐藏状态维度（和 cell 状态维度）：3\n",
    "- 输出维度：2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0b19b7ae-2d1c-466a-939b-9b4696aa9188",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn.functional as F\n",
    "\n",
    "# 设置超参数\n",
    "seq_len = 3\n",
    "batch_size = 3\n",
    "input_size = 4\n",
    "hidden_size = 3\n",
    "output_size = 2\n",
    "\n",
    "# 初始化输入：x 形状为 (seq_len, batch_size, input_size)\n",
    "x = torch.randn(seq_len, batch_size, input_size)\n",
    "\n",
    "# 初始化隐藏状态 h_0 和单元状态 c_0：形状为 (batch_size, hidden_size)\n",
    "h = torch.randn(batch_size, hidden_size)\n",
    "c = torch.randn(batch_size, hidden_size)\n",
    "\n",
    "# 初始化参数\n",
    "def init_param():\n",
    "    return (\n",
    "        torch.randn(hidden_size, input_size),  # W: 对输入 x_t\n",
    "        torch.randn(hidden_size, hidden_size), # U: 对隐藏状态 h_{t-1}\n",
    "        torch.randn(1, hidden_size)            # b: 偏置广播到 batch\n",
    "    )\n",
    "\n",
    "# 初始化各门参数\n",
    "W_f, U_f, b_f = init_param()  # forget gate\n",
    "W_i, U_i, b_i = init_param()  # input gate\n",
    "W_o, U_o, b_o = init_param()  # output gate\n",
    "W_c, U_c, b_c = init_param()  # candidate memory\n",
    "\n",
    "# 输出层参数\n",
    "W_out = torch.randn(output_size, hidden_size)\n",
    "b_out = torch.randn(1, output_size)\n",
    "\n",
    "# 保存每个时间步的输出\n",
    "outputs = []\n",
    "\n",
    "# 开始时间步循环\n",
    "for t in range(seq_len):\n",
    "    x_t = x[t]  # 当前时间步输入：shape = (batch_size, input_size)\n",
    "\n",
    "    # 各门计算\n",
    "    f_t = torch.sigmoid(x_t @ W_f.T + h @ U_f.T + b_f)        # forget gate\n",
    "    i_t = torch.sigmoid(x_t @ W_i.T + h @ U_i.T + b_i)        # input gate\n",
    "    o_t = torch.sigmoid(x_t @ W_o.T + h @ U_o.T + b_o)        # output gate\n",
    "    c_tilde = torch.tanh(x_t @ W_c.T + h @ U_c.T + b_c)       # candidate\n",
    "\n",
    "    # 状态更新\n",
    "    c = f_t * c + i_t * c_tilde                               # new cell state\n",
    "    h = o_t * torch.tanh(c)                                   # new hidden state\n",
    "\n",
    "    # 输出层\n",
    "    y_t = h @ W_out.T + b_out                                 # linear output\n",
    "    outputs.append(y_t)\n",
    "\n",
    "# 拼接输出序列：shape = (seq_len, batch_size, output_size)\n",
    "outputs = torch.stack(outputs, dim=0)\n",
    "\n",
    "print(\"最终输出 outputs.shape:\", outputs.shape)  # 应为 (3, 3, 2)\n",
    "print(\"输出结果:\\n\", outputs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5f67ca88-0098-438d-bc3f-e6721c304f65",
   "metadata": {},
   "source": [
    "### LSTM的简单示例\n",
    "\n",
    "#### 单层单向LSTM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28b9466b-43c6-4f55-a850-cb67ab1d1109",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 参数设置\n",
    "input_size = 10    # 输入特征维度\n",
    "hidden_size = 20   # 隐藏状态维度\n",
    "batch_size = 32    # 批量大小\n",
    "seq_len = 50       # 序列长度\n",
    "\n",
    "# 定义LSTM层\n",
    "lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size, batch_first=True)\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)  # (batch_size, seq_len, input_size)\n",
    "h0 = torch.randn(1, batch_size, hidden_size)          # 初始隐藏状态 (num_layers, batch_size, hidden_size)\n",
    "c0 = torch.randn(1, batch_size, hidden_size)          # 初始记忆单元 (num_layers, batch_size, hidden_size)\n",
    "\n",
    "# 前向传播\n",
    "output, (h_n, c_n) = lstm(input, (h0, c0))\n",
    "\n",
    "# 输出形状\n",
    "print(output.shape)  # (batch_size, seq_len, hidden_size) = (32, 50, 20)\n",
    "print(h_n.shape)    # (num_layers, batch_size, hidden_size) = (1, 32, 20)\n",
    "print(c_n.shape)    # (num_layers, batch_size, hidden_size) = (1, 32, 20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "84974e02-08fe-4661-a0fc-9905082da770",
   "metadata": {},
   "source": [
    "#### 多层双向LSTM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8b273fad-7c1d-4c93-90b1-2f56ea886c36",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 参数设置\n",
    "input_size = 10\n",
    "hidden_size = 20\n",
    "num_layers = 2\n",
    "batch_size = 32\n",
    "seq_len = 50\n",
    "\n",
    "# 定义双向LSTM\n",
    "lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, \n",
    "               batch_first=True, bidirectional=True, dropout=0.3)\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)\n",
    "h0 = torch.randn(2 * 2, batch_size, hidden_size)  # 2层 * 2方向\n",
    "c0 = torch.randn(2 * 2, batch_size, hidden_size)  # 2层 * 2方向\n",
    "\n",
    "# 前向传播\n",
    "output, (h_n, c_n) = lstm(input, (h0, c0))\n",
    "\n",
    "# 输出形状\n",
    "print(output.shape)  # (batch_size, seq_len, num_directions * hidden_size) = (32, 50, 40)\n",
    "print(h_n.shape)    # (num_layers * num_directions, batch_size, hidden_size) = (4, 32, 20)\n",
    "print(c_n.shape)    # (num_layers * num_directions, batch_size, hidden_size) = (4, 32, 20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c56604d5-285d-4e41-aa6d-a856002a098e",
   "metadata": {},
   "source": [
    "#### 处理变长序列"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6510fa2d-b866-43e4-8507-183532a4c39a",
   "metadata": {},
   "outputs": [],
   "source": [
    "from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence\n",
    "\n",
    "# 模拟变长序列\n",
    "input = torch.randn(batch_size, seq_len, input_size)\n",
    "lengths = torch.randint(10, seq_len + 1, (batch_size,))  # 每个序列的实际长度\n",
    "\n",
    "# 按长度降序排序\n",
    "lengths, indices = lengths.sort(descending=True)\n",
    "input = input[indices]\n",
    "\n",
    "# 定义LSTM\n",
    "lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size, batch_first=True)\n",
    "\n",
    "# 打包序列\n",
    "packed_input = pack_padded_sequence(input, lengths, batch_first=True, enforce_sorted=True)\n",
    "\n",
    "# 前向传播\n",
    "output, (h_n, c_n) = lstm(packed_input)\n",
    "\n",
    "# 解包输出\n",
    "output, output_lengths = pad_packed_sequence(output, batch_first=True)\n",
    "\n",
    "print(output.shape)  # (batch_size, seq_len, hidden_size)\n",
    "print(h_n.shape)    # (1, batch_size, hidden_size)\n",
    "print(c_n.shape)    # (1, batch_size, hidden_size)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d4c430e4-a483-4ce3-96fb-33e1c9ac6a6c",
   "metadata": {},
   "source": [
    "#### 权重访问\n",
    "\n",
    "- LSTM层的权重矩阵可以通过 lstm.weight_ih_l[k]（输入到隐藏层的权重）和 lstm.weight_hh_l[k]（隐藏层到隐藏层的权重）访问，其中 k 是层索引。\n",
    "- 偏置项通过 lstm.bias_ih_l[k] 和 lstm.bias_hh_l[k] 访问。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "77ad3ba3-1909-4123-b601-129b331161fb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 访问第一层LSTM的权重\n",
    "print(lstm.weight_ih_l0.shape)  # (4 * hidden_size, input_size)，4个权重矩阵（输入门、遗忘门、输出门、候选记忆单元）\n",
    "print(lstm.weight_hh_l0.shape)  # (4 * hidden_size, hidden_size)\n",
    "print(lstm.bias_ih_l0.shape)    # (4 * hidden_size,)\n",
    "print(lstm.bias_hh_l0.shape)    # (4 * hidden_size,)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c05fcafd-3f89-4d7c-862f-843d76d08de6",
   "metadata": {},
   "source": [
    "### 基于IMDB数据集的LSTM模型实现示例\n",
    "\n",
    "提示：需要事先安装依赖的模块。重新安装，以确保几个模块间版本上的兼容：\n",
    "```bash\n",
    "pip install --force-reinstall numpy==2.1.3 tensorflow gensim scipy\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "497f4a14-9c0b-4db6-938e-16873e0ed5d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from tensorflow.keras.datasets import imdb\n",
    "import numpy as np\n",
    "import re\n",
    "import string\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "np.random.seed(42)\n",
    "\n",
    "# 1. 数据集准备\n",
    "class IMDBDataset(Dataset):\n",
    "    def __init__(self, data, labels, max_len=500):\n",
    "        self.data = data\n",
    "        self.labels = labels\n",
    "        self.max_len = max_len\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.data)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        # 截断或填充序列到 max_len\n",
    "        seq = self.data[idx][:self.max_len]\n",
    "        seq = np.pad(seq, (0, self.max_len - len(seq)), mode='constant') if len(seq) < self.max_len else seq\n",
    "        return torch.tensor(seq, dtype=torch.long), torch.tensor(self.labels[idx], dtype=torch.float)\n",
    "\n",
    "# 加载IMDB数据集和词典\n",
    "vocab_size = 20000  # 限制词汇表大小\n",
    "max_len = 500       # 最大序列长度\n",
    "(x_train, y_train), (x_test, y_test) = imdb.load_data(num_words=vocab_size)\n",
    "word_index = imdb.get_word_index()  # 获取单词到索引的映射\n",
    "\n",
    "# 创建PyTorch数据集\n",
    "train_dataset = IMDBDataset(x_train, y_train, max_len)\n",
    "test_dataset = IMDBDataset(x_test, y_test, max_len)\n",
    "\n",
    "# 创建数据加载器\n",
    "batch_size = 32\n",
    "train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)\n",
    "test_loader = DataLoader(test_dataset, batch_size=batch_size)\n",
    "\n",
    "# 2. 定义LSTM模型\n",
    "class IMDBLSTM(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size, output_size, num_layers=1):\n",
    "        super(IMDBLSTM, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)  # 词嵌入层\n",
    "        self.lstm = nn.LSTM(\n",
    "            input_size=embed_size,\n",
    "            hidden_size=hidden_size,\n",
    "            num_layers=num_layers,\n",
    "            batch_first=True\n",
    "        )\n",
    "        self.fc = nn.Linear(hidden_size, output_size)  # 输出层\n",
    "        self.sigmoid = nn.Sigmoid()  # 用于二分类\n",
    "\n",
    "    def forward(self, x, h_0=None, c_0=None):\n",
    "        # x: [batch_size, seq_len]\n",
    "        x = self.embedding(x)  # [batch_size, seq_len, embed_size]\n",
    "        # LSTM需要h_0和c_0，若未提供则初始化为零\n",
    "        if h_0 is None:\n",
    "            h_0 = torch.zeros(self.lstm.num_layers, x.size(0), self.lstm.hidden_size).to(x.device)\n",
    "        if c_0 is None:\n",
    "            c_0 = torch.zeros(self.lstm.num_layers, x.size(0), self.lstm.hidden_size).to(x.device)\n",
    "        output, (h_n, c_n) = self.lstm(x, (h_0, c_0))  # output: [batch_size, seq_len, hidden_size]\n",
    "        out = self.fc(output[:, -1, :])  # 取最后一个时间步: [batch_size, output_size]\n",
    "        out = self.sigmoid(out)  # [batch_size, 1]\n",
    "        return out\n",
    "\n",
    "# 3. 模型参数\n",
    "embed_size = 128    # 词嵌入维度\n",
    "hidden_size = 256   # 隐藏状态维度\n",
    "output_size = 1     # 输出维度（二分类）\n",
    "num_layers = 1      # LSTM层数\n",
    "\n",
    "# 初始化模型、损失函数和优化器\n",
    "model = IMDBLSTM(vocab_size, embed_size, hidden_size, output_size, num_layers)\n",
    "criterion = nn.BCELoss()  # 二分类交叉熵损失\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 4. 训练模型\n",
    "def train_model(model, train_loader, criterion, optimizer, epochs=5, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for data, labels in train_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(data)\n",
    "            loss = criterion(output, labels)\n",
    "            loss.backward()\n",
    "            # 梯度裁剪防止爆炸\n",
    "            #torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1)\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f\"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}\")\n",
    "\n",
    "# 5. 评估模型\n",
    "def evaluate_model(model, test_loader, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for data, labels in test_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            output = model(data)\n",
    "            predictions = (output >= 0.5).float()\n",
    "            correct += (predictions == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    accuracy = correct / total * 100\n",
    "    print(f\"Test Accuracy: {accuracy:.2f}%\")\n",
    "\n",
    "# 6. 测试功能：处理示例评论\n",
    "def text_to_sequence(text, word_index, max_len=500):\n",
    "    # 文本预处理：小写、去除标点\n",
    "    text = text.lower()\n",
    "    text = re.sub(f'[{string.punctuation}]', '', text)\n",
    "    # 分词\n",
    "    tokens = text.split()\n",
    "    # 转换为单词索引\n",
    "    sequence = [word_index.get(word, 0) for word in tokens if word_index.get(word, 0) < vocab_size]\n",
    "    # 截断或填充\n",
    "    sequence = sequence[:max_len]\n",
    "    sequence = np.pad(sequence, (0, max_len - len(sequence)), mode='constant') if len(sequence) < max_len else sequence\n",
    "    return torch.tensor([sequence], dtype=torch.long)  # [1, max_len]\n",
    "\n",
    "def test_model(model, texts, word_index, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        for text in texts:\n",
    "            # 转换为序列\n",
    "            seq = text_to_sequence(text, word_index, max_len)\n",
    "            seq = seq.to(device)\n",
    "            # 预测\n",
    "            output = model(seq)  # [1, 1]\n",
    "            prob = output.item()\n",
    "            sentiment = \"Positive\" if prob >= 0.5 else \"Negative\"\n",
    "            print(f\"Text: {text}\")\n",
    "            print(f\"Sentiment: {sentiment}, Probability: {prob:.4f}\\n\")\n",
    "\n",
    "# 7. 运行训练、评估和测试\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "train_model(model, train_loader, criterion, optimizer, epochs=10, device=device)\n",
    "evaluate_model(model, test_loader, device=device)\n",
    "\n",
    "# 示例评论\n",
    "test_texts = [\n",
    "    \"I absolutely love this movie, it's fantastic and thrilling!\",\n",
    "    \"This film was boring and a complete waste of time.\",\n",
    "    \"The plot was okay, but the acting was amazing.\",\n",
    "    \"Terrible movie, I hated every minute of it.\"\n",
    "]\n",
    "test_model(model, test_texts, word_index, device=device)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "088a2482-826c-418f-b5bf-84da7115deb8",
   "metadata": {},
   "source": [
    "### 扩展上面的LSTM示例\n",
    "\n",
    "提示：更新相关的包至最新版本。\n",
    "\n",
    "```bash\n",
    "pip install --upgrade gensim numpy scipy torch tensorflow\n",
    "```\n",
    "\n",
    "\n",
    "**改进说明**\n",
    "\n",
    "1. 双向LSTM：\n",
    "- 设置 nn.LSTM(bidirectional=True)，使模型同时从正向和反向处理序列。\n",
    "- 输出维度变为 hidden_size * 2（正向和反向隐藏状态拼接）。\n",
    "- 调整全连接层输入维度，适应双向输出。\n",
    "\n",
    "2. Dropout：\n",
    "- 在LSTM层设置dropout=0.5，在多层LSTM的隐藏层之间应用Dropout（不影响最后一层）。\n",
    "- 在全连接层后添加nn.Dropout(0.5)，进一步正则化。\n",
    "\n",
    "3. 超参数调优：\n",
    "- 隐藏状态维度：hidden_size=512（增大以提高模型容量）。\n",
    "- 层数：num_layers=2（堆叠两层LSTM，增强表达能力）。\n",
    "- 训练轮数：epochs=10（增加以确保充分训练）。\n",
    "- 学习率：测试 lr=0.0005（稍降低以稳定训练）。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc5426c8-7cd1-44c5-9140-75602a6b040e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from tensorflow.keras.datasets import imdb\n",
    "import numpy as np\n",
    "import re\n",
    "import string\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "np.random.seed(42)\n",
    "\n",
    "# 1. 数据集准备\n",
    "class IMDBDataset(Dataset):\n",
    "    def __init__(self, data, labels, max_len=500):\n",
    "        self.data = data\n",
    "        self.labels = labels\n",
    "        self.max_len = max_len\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.data)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        # 截断或填充序列到 max_len\n",
    "        seq = self.data[idx][:self.max_len]\n",
    "        seq = np.pad(seq, (0, self.max_len - len(seq)), mode='constant') if len(seq) < self.max_len else seq\n",
    "        return torch.tensor(seq, dtype=torch.long), torch.tensor(self.labels[idx], dtype=torch.float)\n",
    "\n",
    "# 加载IMDB数据集和词典\n",
    "vocab_size = 20000  # 限制词汇表大小\n",
    "max_len = 500       # 最大序列长度\n",
    "(x_train, y_train), (x_test, y_test) = imdb.load_data(num_words=vocab_size)\n",
    "word_index = imdb.get_word_index()  # 获取单词到索引的映射\n",
    "\n",
    "# 创建PyTorch数据集\n",
    "train_dataset = IMDBDataset(x_train, y_train, max_len)\n",
    "test_dataset = IMDBDataset(x_test, y_test, max_len)\n",
    "\n",
    "# 创建数据加载器\n",
    "batch_size = 32\n",
    "train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)\n",
    "test_loader = DataLoader(test_dataset, batch_size=batch_size)\n",
    "\n",
    "# 2. 初始化嵌入矩阵（随机初始化）\n",
    "embed_size = 300  # 嵌入维度\n",
    "embedding_matrix = torch.randn(vocab_size, embed_size)  # 随机初始化嵌入矩阵\n",
    "\n",
    "# 3. 定义改进的LSTM模型\n",
    "class IMDBLSTM(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size, output_size, num_layers=2):\n",
    "        super(IMDBLSTM, self).__init__()\n",
    "        # 初始化嵌入层并加载随机权重\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.embedding.weight = nn.Parameter(embedding_matrix)\n",
    "        self.embedding.weight.requires_grad = True  # 允许微调嵌入权重\n",
    "        \n",
    "        # 双向LSTM，设置dropout=0.5\n",
    "        self.lstm = nn.LSTM(\n",
    "            input_size=embed_size,\n",
    "            hidden_size=hidden_size,\n",
    "            num_layers=num_layers,\n",
    "            batch_first=True,\n",
    "            bidirectional=True,  # 启用双向LSTM\n",
    "            dropout=0.5 if num_layers > 1 else 0  # 多层LSTM间应用Dropout\n",
    "        )\n",
    "        # 全连接层，输入维度为hidden_size * 2（双向LSTM）\n",
    "        self.fc = nn.Linear(hidden_size * 2, output_size)\n",
    "        self.dropout = nn.Dropout(0.5)  # 全连接层后的Dropout\n",
    "        self.sigmoid = nn.Sigmoid()  # 用于二分类\n",
    "\n",
    "    def forward(self, x, h_0=None, c_0=None):\n",
    "        # x: [batch_size, seq_len]\n",
    "        x = self.embedding(x)  # [batch_size, seq_len, embed_size]\n",
    "        # 初始化隐藏状态和单元状态\n",
    "        if h_0 is None:\n",
    "            h_0 = torch.zeros(self.lstm.num_layers * 2, x.size(0), self.lstm.hidden_size).to(x.device)  # 双向需乘2\n",
    "        if c_0 is None:\n",
    "            c_0 = torch.zeros(self.lstm.num_layers * 2, x.size(0), self.lstm.hidden_size).to(x.device)\n",
    "        # LSTM输出: [batch_size, seq_len, hidden_size * 2]\n",
    "        output, (h_n, c_n) = self.lstm(x, (h_0, c_0))\n",
    "        # 取最后一个时间步，拼接正向和反向隐藏状态\n",
    "        out = self.fc(self.dropout(output[:, -1, :]))  # [batch_size, output_size]\n",
    "        out = self.sigmoid(out)  # [batch_size, 1]\n",
    "        return out\n",
    "\n",
    "# 4. 模型参数\n",
    "hidden_size = 512   # 增大隐藏状态维度\n",
    "output_size = 1     # 输出维度（二分类）\n",
    "num_layers = 2      # 堆叠两层LSTM\n",
    "\n",
    "# 初始化模型、损失函数和优化器\n",
    "model = IMDBLSTM(vocab_size, embed_size, hidden_size, output_size, num_layers)\n",
    "criterion = nn.BCELoss()  # 二分类交叉熵损失\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.0005)  # 降低学习率以稳定训练\n",
    "\n",
    "# 5. 训练模型\n",
    "def train_model(model, train_loader, criterion, optimizer, epochs=10, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for data, labels in train_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(data)\n",
    "            loss = criterion(output, labels)\n",
    "            loss.backward()\n",
    "            # 梯度裁剪防止爆炸\n",
    "            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1)\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f\"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}\")\n",
    "\n",
    "# 6. 评估模型\n",
    "def evaluate_model(model, test_loader, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for data, labels in test_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            output = model(data)\n",
    "            predictions = (output >= 0.5).float()\n",
    "            correct += (predictions == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    accuracy = correct / total * 100\n",
    "    print(f\"Test Accuracy: {accuracy:.2f}%\")\n",
    "\n",
    "# 7. 测试功能：处理示例评论\n",
    "def text_to_sequence(text, word_index, max_len=500):\n",
    "    # 文本预处理：小写、去除标点\n",
    "    text = text.lower()\n",
    "    text = re.sub(f'[{string.punctuation}]', '', text)\n",
    "    # 分词\n",
    "    tokens = text.split()\n",
    "    # 转换为单词索引\n",
    "    sequence = [word_index.get(word, 0) for word in tokens if word_index.get(word, 0) < vocab_size]\n",
    "    # 截断或填充\n",
    "    sequence = sequence[:max_len]\n",
    "    sequence = np.pad(sequence, (0, max_len - len(sequence)), mode='constant') if len(sequence) < max_len else sequence\n",
    "    return torch.tensor([sequence], dtype=torch.long)  # [1, max_len]\n",
    "\n",
    "def test_model(model, texts, word_index, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        for text in texts:\n",
    "            # 转换为序列\n",
    "            seq = text_to_sequence(text, word_index, max_len)\n",
    "            seq = seq.to(device)\n",
    "            # 预测\n",
    "            output = model(seq)  # [1, 1]\n",
    "            prob = output.item()\n",
    "            sentiment = \"Positive\" if prob >= 0.5 else \"Negative\"\n",
    "            print(f\"Text: {text}\")\n",
    "            print(f\"Sentiment: {sentiment}, Probability: {prob:.4f}\\n\")\n",
    "\n",
    "# 8. 运行训练、评估和测试\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "train_model(model, train_loader, criterion, optimizer, epochs=10, device=device)\n",
    "evaluate_model(model, test_loader, device=device)\n",
    "\n",
    "# 示例评论\n",
    "test_texts = [\n",
    "    \"I absolutely love this movie, it's fantastic and thrilling!\",\n",
    "    \"This film was boring and a complete waste of time.\",\n",
    "    \"The plot was okay, but the acting was amazing.\",\n",
    "    \"Terrible movie, I hated every minute of it.\"\n",
    "]\n",
    "test_model(model, test_texts, word_index, device=device)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7533a9cb-acc6-49af-bf34-4d22fc95d07b",
   "metadata": {},
   "source": [
    "### RNN BPTT示例"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff51bdb6-79d0-41f1-ac2f-fe1ee31ef166",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "def tanh(x):\n",
    "    return np.tanh(x)\n",
    "\n",
    "def tanh_deriv(x):\n",
    "    return 1 - np.tanh(x) ** 2\n",
    "\n",
    "# 参数设置\n",
    "input_dim, hidden_dim, output_dim = 10, 20, 5\n",
    "W_xh = np.random.randn(hidden_dim, input_dim)\n",
    "W_hh = np.random.randn(hidden_dim, hidden_dim)\n",
    "W_hy = np.random.randn(output_dim, hidden_dim)\n",
    "b_h = np.random.randn(hidden_dim)\n",
    "b_y = np.random.randn(output_dim)\n",
    "\n",
    "def rnn_forward(sequence, h_0):\n",
    "    h_t = h_0\n",
    "    hidden_states = [h_t]\n",
    "    outputs = []\n",
    "    zs = []  # 存储 z_t = W_hh * h_{t-1} + W_xh * x_t + b_h\n",
    "    for x_t in sequence:\n",
    "        z_t = np.dot(W_hh, h_t) + np.dot(W_xh, x_t) + b_h\n",
    "        h_t = tanh(z_t)\n",
    "        y_t = np.dot(W_hy, h_t) + b_y\n",
    "        hidden_states.append(h_t)\n",
    "        outputs.append(y_t)\n",
    "        zs.append(z_t)\n",
    "    return hidden_states, outputs, zs\n",
    "\n",
    "def rnn_backward(sequence, targets, hidden_states, outputs, zs):\n",
    "    T = len(sequence)\n",
    "    dW_xh = np.zeros_like(W_xh)\n",
    "    dW_hh = np.zeros_like(W_hh)\n",
    "    dW_hy = np.zeros_like(W_hy)\n",
    "    db_h = np.zeros_like(b_h)\n",
    "    db_y = np.zeros_like(b_y)\n",
    "    dh_next = np.zeros_like(hidden_states[0])\n",
    "\n",
    "    for t in range(T-1, -1, -1):\n",
    "        # 损失对 \\hat{y}_t 的导数\n",
    "        dy_t = outputs[t] - targets[t]\n",
    "        # 对 W_hy 和 b_y 的梯度\n",
    "        dW_hy += np.outer(dy_t, hidden_states[t+1])\n",
    "        db_y += dy_t\n",
    "        # 对 h_t 的梯度\n",
    "        dh_t = np.dot(W_hy.T, dy_t) + dh_next\n",
    "        # 对 z_t 的梯度\n",
    "        dz_t = dh_t * tanh_deriv(zs[t])\n",
    "        # 对 W_hh, W_xh, b_h 的梯度\n",
    "        dW_hh += np.outer(dz_t, hidden_states[t])\n",
    "        dW_xh += np.outer(dz_t, sequence[t])\n",
    "        db_h += dz_t\n",
    "        # 传递到上一时间步\n",
    "        dh_next = np.dot(W_hh.T, dz_t)\n",
    "\n",
    "    return dW_xh, dW_hh, dW_hy, db_h, db_y\n",
    "\n",
    "# 示例调用\n",
    "sequence = [np.random.randn(input_dim) for _ in range(3)]\n",
    "targets = [np.random.randn(output_dim) for _ in range(3)]\n",
    "h_0 = np.zeros(hidden_dim)\n",
    "hidden_states, outputs, zs = rnn_forward(sequence, h_0)\n",
    "grads = rnn_backward(sequence, targets, hidden_states, outputs, zs)\n",
    "print(\"Gradients:\", grads)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "65522034-75a9-40e4-a61b-90011611fdb1",
   "metadata": {},
   "source": [
    "## GRU示例\n",
    "\n",
    "### 手动实现GRU前向传播的计算过程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "482f96c3-c3bd-45e2-b055-44eb53be10ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn.functional as F\n",
    "\n",
    "# 假设输入维度和隐藏状态维度\n",
    "input_size = 4\n",
    "hidden_size = 3\n",
    "output_size = 2  # 最终输出层维度\n",
    "\n",
    "# 模拟一个输入向量x_t 和前一时刻隐藏状态h_{t-1}\n",
    "x_t = torch.randn(input_size)       # 输入 (4,)\n",
    "h_prev = torch.randn(hidden_size)   # h_{t-1} (3,)\n",
    "\n",
    "# 初始化参数：权重和偏置\n",
    "# 重置门参数\n",
    "W_r = torch.randn(hidden_size, input_size)\n",
    "U_r = torch.randn(hidden_size, hidden_size)\n",
    "b_r = torch.randn(hidden_size)\n",
    "\n",
    "# 更新门参数\n",
    "W_z = torch.randn(hidden_size, input_size)\n",
    "U_z = torch.randn(hidden_size, hidden_size)\n",
    "b_z = torch.randn(hidden_size)\n",
    "\n",
    "# 候选隐藏状态参数\n",
    "W_h = torch.randn(hidden_size, input_size)\n",
    "U_h = torch.randn(hidden_size, hidden_size)\n",
    "b_h = torch.randn(hidden_size)\n",
    "\n",
    "# 最终输出层参数\n",
    "W_out = torch.randn(output_size, hidden_size)\n",
    "b_out = torch.randn(output_size)\n",
    "\n",
    "# ========================\n",
    "# 1. 重置门 r_t\n",
    "# ========================\n",
    "r_t = torch.sigmoid(W_r @ x_t + U_r @ h_prev + b_r)\n",
    "print(\"重置门 r_t:\", r_t)\n",
    "\n",
    "# ========================\n",
    "# 2. 更新门 z_t\n",
    "# ========================\n",
    "z_t = torch.sigmoid(W_z @ x_t + U_z @ h_prev + b_z)\n",
    "print(\"更新门 z_t:\", z_t)\n",
    "\n",
    "# ========================\n",
    "# 3. 候选隐藏状态 h_t~\n",
    "# ========================\n",
    "h_tilde = torch.tanh(W_h @ x_t + U_h @ (r_t * h_prev) + b_h)\n",
    "print(\"候选隐藏状态 h~_t:\", h_tilde)\n",
    "\n",
    "# ========================\n",
    "# 4. 当前隐藏状态 h_t\n",
    "# ========================\n",
    "h_t = (1 - z_t) * h_prev + z_t * h_tilde\n",
    "print(\"最终隐藏状态 h_t:\", h_t)\n",
    "\n",
    "# ========================\n",
    "# 5. 输出层（例如分类任务）\n",
    "# ========================\n",
    "y_t = W_out @ h_t + b_out\n",
    "print(\"预测输出 y_t:\", y_t)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c624260d-e635-43b7-9263-fbbb6ccf509d",
   "metadata": {},
   "source": [
    "### 多时间步，支持batch的GRU前向传播\n",
    "\n",
    "**设置说明：**\n",
    "- 输入序列长度：3（即3个时间步）\n",
    "- Batch大小：3\n",
    "- 每个时间步输入维度：4\n",
    "- 隐藏状态维度：3\n",
    "- 输出维度：2（用于预测）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "845330fd-83df-4a3e-be05-7485c3a642e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn.functional as F\n",
    "\n",
    "# 设置参数\n",
    "seq_len = 3\n",
    "batch_size = 3\n",
    "input_size = 4\n",
    "hidden_size = 3\n",
    "output_size = 2\n",
    "\n",
    "# 模拟一个输入序列: (seq_len, batch_size, input_size)\n",
    "x = torch.randn(seq_len, batch_size, input_size)\n",
    "\n",
    "# 初始化隐藏状态 h_0: (batch_size, hidden_size)\n",
    "h = torch.randn(batch_size, hidden_size)\n",
    "\n",
    "# 初始化参数\n",
    "def init_param(*shape):\n",
    "    return torch.randn(*shape)\n",
    "\n",
    "# 重置门参数\n",
    "W_r = init_param(hidden_size, input_size)\n",
    "U_r = init_param(hidden_size, hidden_size)\n",
    "b_r = init_param(1, hidden_size)\n",
    "\n",
    "# 更新门参数\n",
    "W_z = init_param(hidden_size, input_size)\n",
    "U_z = init_param(hidden_size, hidden_size)\n",
    "b_z = init_param(1, hidden_size)\n",
    "\n",
    "# 候选隐藏状态参数\n",
    "W_h = init_param(hidden_size, input_size)\n",
    "U_h = init_param(hidden_size, hidden_size)\n",
    "b_h = init_param(1, hidden_size)\n",
    "\n",
    "# 输出层参数\n",
    "W_out = init_param(output_size, hidden_size)\n",
    "b_out = init_param(1, output_size)\n",
    "\n",
    "# 保存每个时间步的输出\n",
    "outputs = []\n",
    "\n",
    "for t in range(seq_len):\n",
    "    x_t = x[t]  # (batch_size, input_size)\n",
    "    \n",
    "    # 重置门 r_t: (batch_size, hidden_size)\n",
    "    r_t = torch.sigmoid(x_t @ W_r.T + h @ U_r.T + b_r)\n",
    "    \n",
    "    # 更新门 z_t: (batch_size, hidden_size)\n",
    "    z_t = torch.sigmoid(x_t @ W_z.T + h @ U_z.T + b_z)\n",
    "    \n",
    "    # 候选隐藏状态 h~_t: (batch_size, hidden_size)\n",
    "    h_tilde = torch.tanh(x_t @ W_h.T + (r_t * h) @ U_h.T + b_h)\n",
    "    \n",
    "    # 当前隐藏状态 h_t: (batch_size, hidden_size)\n",
    "    h = (1 - z_t) * h + z_t * h_tilde\n",
    "    \n",
    "    # 输出层 y_t: (batch_size, output_size)\n",
    "    y_t = h @ W_out.T + b_out  # Linear output layer\n",
    "    outputs.append(y_t)\n",
    "\n",
    "# 堆叠输出序列: (seq_len, batch_size, output_size)\n",
    "outputs = torch.stack(outputs, dim=0)\n",
    "\n",
    "print(\"初始输入 x_0:\\n\", x)\n",
    "print(\"初始隐藏状态h_0:\\n\", h)\n",
    "print(\"最终输出 outputs.shape:\", outputs.shape)  # 应为 (3, 3, 2)\n",
    "print(\"输出结果:\\n\", outputs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "63094fc6-508a-4524-bfed-b94934001a64",
   "metadata": {},
   "source": [
    "### GRU的简单使用示例\n",
    "\n",
    "**单层单向GRU示例**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4cccaf2c-7caf-48b1-8a36-bec1da29e172",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 参数设置\n",
    "input_size = 10    # 输入特征维度\n",
    "hidden_size = 20   # 隐藏状态维度\n",
    "batch_size = 32    # 批量大小\n",
    "seq_len = 50       # 序列长度\n",
    "\n",
    "# 定义GRU层\n",
    "gru = nn.GRU(input_size=input_size, hidden_size=hidden_size, batch_first=True)\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)  # (batch_size, seq_len, input_size)\n",
    "h0 = torch.randn(1, batch_size, hidden_size)          # 初始隐藏状态 (num_layers, batch_size, hidden_size)\n",
    "\n",
    "# 前向传播\n",
    "output, h_n = gru(input, h0)\n",
    "\n",
    "# 输出形状\n",
    "print(output.shape)  # (batch_size, seq_len, hidden_size) = (32, 50, 20)\n",
    "print(h_n.shape)    # (num_layers, batch_size, hidden_size) = (1, 32, 20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "df983da8-d268-4d51-9da9-7428bebc6eb8",
   "metadata": {},
   "source": [
    "#### 多层双向GRU示例"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e2734521-6b98-4de2-bcd6-9c184872385c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 参数设置\n",
    "input_size = 10\n",
    "hidden_size = 20\n",
    "num_layers = 2\n",
    "batch_size = 32\n",
    "seq_len = 50\n",
    "\n",
    "# 定义双向GRU\n",
    "gru = nn.GRU(input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, \n",
    "             batch_first=True, bidirectional=True, dropout=0.3)\n",
    "\n",
    "# 输入数据\n",
    "input = torch.randn(batch_size, seq_len, input_size)\n",
    "h0 = torch.randn(2 * 2, batch_size, hidden_size)  # 2层 * 2方向\n",
    "\n",
    "# 前向传播\n",
    "output, h_n = gru(input, h0)\n",
    "\n",
    "# 输出形状\n",
    "print(output.shape)  # (batch_size, seq_len, num_directions * hidden_size) = (32, 50, 40)\n",
    "print(h_n.shape)    # (num_layers * num_directions, batch_size, hidden_size) = (4, 32, 20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a68b1203-8544-4f29-854f-8f89791edf88",
   "metadata": {},
   "source": [
    "#### 处理变长输入序列"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a43d9dc-af50-43a5-aaa8-052e74b45469",
   "metadata": {},
   "outputs": [],
   "source": [
    "from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence\n",
    "\n",
    "# 模拟变长序列\n",
    "input = torch.randn(batch_size, seq_len, input_size)\n",
    "lengths = torch.randint(10, seq_len + 1, (batch_size,))  # 每个序列的实际长度\n",
    "\n",
    "# 按长度降序排序（必要步骤）\n",
    "lengths, indices = lengths.sort(descending=True)\n",
    "input = input[indices]\n",
    "\n",
    "# 定义GRU\n",
    "gru = nn.GRU(input_size=input_size, hidden_size=hidden_size, batch_first=True)\n",
    "\n",
    "# 打包序列\n",
    "packed_input = pack_padded_sequence(input, lengths, batch_first=True, enforce_sorted=True)\n",
    "\n",
    "# 前向传播\n",
    "output, h_n = gru(packed_input)\n",
    "\n",
    "# 解包输出\n",
    "output, output_lengths = pad_packed_sequence(output, batch_first=True)\n",
    "\n",
    "print(output.shape)  # (batch_size, seq_len, hidden_size)\n",
    "print(h_n.shape)    # (1, batch_size, hidden_size)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "47c7261b-30f3-4227-8ba2-8b5bebc63b53",
   "metadata": {},
   "source": [
    "### GRU模型案例\n",
    "\n",
    "修改前面LSTM示例使用GRU层实现：使用PyTorch实现一个简单的GRU模型，完成IMDB电影评论的情感分析任务（二分类：正面或负面）。下面的代码包括数据加载、预处理、模型定义、训练和评估等功能的定义。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d1f76ef7-c2d1-4a1f-b1a5-389dfc34be83",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from tensorflow.keras.datasets import imdb\n",
    "import numpy as np\n",
    "import re\n",
    "import string\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "np.random.seed(42)\n",
    "\n",
    "# 1. 数据集准备\n",
    "class IMDBDataset(Dataset):\n",
    "    def __init__(self, data, labels, max_len=500):\n",
    "        self.data = data\n",
    "        self.labels = labels\n",
    "        self.max_len = max_len\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.data)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        # 截断或填充序列到 max_len\n",
    "        seq = self.data[idx][:self.max_len]\n",
    "        seq = np.pad(seq, (0, self.max_len - len(seq)), mode='constant') if len(seq) < self.max_len else seq\n",
    "        return torch.tensor(seq, dtype=torch.long), torch.tensor(self.labels[idx], dtype=torch.float)\n",
    "\n",
    "# 加载IMDB数据集和词典\n",
    "vocab_size = 20000  # 限制词汇表大小\n",
    "max_len = 500       # 最大序列长度\n",
    "(x_train, y_train), (x_test, y_test) = imdb.load_data(num_words=vocab_size)\n",
    "word_index = imdb.get_word_index()  # 获取单词到索引的映射\n",
    "\n",
    "# 创建PyTorch数据集\n",
    "train_dataset = IMDBDataset(x_train, y_train, max_len)\n",
    "test_dataset = IMDBDataset(x_test, y_test, max_len)\n",
    "\n",
    "# 创建数据加载器\n",
    "batch_size = 32\n",
    "train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)\n",
    "test_loader = DataLoader(test_dataset, batch_size=batch_size)\n",
    "\n",
    "# 2. 定义GRU模型\n",
    "# 关键修改点：\n",
    "# - 类名从IMDBLSTM变为IMDBGRU\n",
    "# - 将 nn.LSTM替换为nn.GRU\n",
    "# - forward方法中，GRU的输出元组只有一个隐藏状态h_n，不需要c_n\n",
    "class IMDBGRU(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size, output_size, num_layers=1):\n",
    "        super(IMDBGRU, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.gru = nn.GRU( # <-- 修改点\n",
    "            input_size=embed_size,\n",
    "            hidden_size=hidden_size,\n",
    "            num_layers=num_layers,\n",
    "            batch_first=True\n",
    "        )\n",
    "        self.fc = nn.Linear(hidden_size, output_size)\n",
    "        self.sigmoid = nn.Sigmoid()\n",
    "\n",
    "    def forward(self, x, h_0=None):\n",
    "        # x: [batch_size, seq_len]\n",
    "        x = self.embedding(x) # [batch_size, seq_len, embed_size]\n",
    "        \n",
    "        # GRU只需要隐藏状态h_0，若未提供则初始化为零\n",
    "        # h_0 的形状为 (num_layers * num_directions, batch, hidden_size)\n",
    "        if h_0 is None:\n",
    "            h_0 = torch.zeros(self.gru.num_layers, x.size(0), self.gru.hidden_size).to(x.device)\n",
    "            \n",
    "        output, h_n = self.gru(x, h_0) # <-- 修改点，GRU的第二个输出是h_n，没有c_n\n",
    "        \n",
    "        # h_n 是最后一个时间步的隐藏状态，形状为[num_layers, batch_size, hidden_size]\n",
    "        # 取最后一个时间步的隐藏状态作为分类器的输入\n",
    "        out = self.fc(h_n[-1, :, :]) # <-- 修改点，这里可以直接用h_n[-1, :, :]\n",
    "        out = self.sigmoid(out)\n",
    "        return out\n",
    "\n",
    "# 3. 模型参数\n",
    "embed_size = 128    # 词嵌入维度\n",
    "hidden_size = 256   # 隐藏状态维度\n",
    "output_size = 1     # 输出维度（二分类）\n",
    "num_layers = 1      # GRU层数\n",
    "\n",
    "# 初始化模型、损失函数和优化器\n",
    "model = IMDBGRU(vocab_size, embed_size, hidden_size, output_size, num_layers) # <-- 修改点\n",
    "criterion = nn.BCELoss()  # 二分类交叉熵损失\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 4. 训练模型\n",
    "def train_model(model, train_loader, criterion, optimizer, epochs=10, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for data, labels in train_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(data)\n",
    "            loss = criterion(output, labels)\n",
    "            loss.backward()\n",
    "            # 梯度裁剪防止爆炸\n",
    "            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1)\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f\"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}\")\n",
    "\n",
    "# 5. 评估模型\n",
    "def evaluate_model(model, test_loader, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for data, labels in test_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            output = model(data)\n",
    "            predictions = (output >= 0.5).float()\n",
    "            correct += (predictions == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    accuracy = correct / total * 100\n",
    "    print(f\"Test Accuracy: {accuracy:.2f}%\")\n",
    "\n",
    "# 6. 测试功能：处理示例评论\n",
    "def text_to_sequence(text, word_index, max_len=500):\n",
    "    # 文本预处理：小写、去除标点\n",
    "    text = text.lower()\n",
    "    text = re.sub(f'[{string.punctuation}]', '', text)\n",
    "    # 分词\n",
    "    tokens = text.split()\n",
    "    # 转换为单词索引\n",
    "    sequence = [word_index.get(word, 0) for word in tokens if word_index.get(word, 0) < vocab_size]\n",
    "    # 截断或填充\n",
    "    sequence = sequence[:max_len]\n",
    "    sequence = np.pad(sequence, (0, max_len - len(sequence)), mode='constant') if len(sequence) < max_len else sequence\n",
    "    return torch.tensor([sequence], dtype=torch.long)  # [1, max_len]\n",
    "\n",
    "def test_model(model, texts, word_index, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        for text in texts:\n",
    "            # 转换为序列\n",
    "            seq = text_to_sequence(text, word_index, max_len)\n",
    "            seq = seq.to(device)\n",
    "            # 预测\n",
    "            output = model(seq)  # [1, 1]\n",
    "            prob = output.item()\n",
    "            sentiment = \"Positive\" if prob >= 0.5 else \"Negative\"\n",
    "            print(f\"Text: {text}\")\n",
    "            print(f\"Sentiment: {sentiment}, Probability: {prob:.4f}\\n\")\n",
    "\n",
    "# 7. 运行训练、评估和测试\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "train_model(model, train_loader, criterion, optimizer, epochs=10, device=device)\n",
    "evaluate_model(model, test_loader, device=device)\n",
    "\n",
    "# 示例评论\n",
    "test_texts = [\n",
    "    \"I absolutely love this movie, it's fantastic and thrilling!\",\n",
    "    \"This film was boring and a complete waste of time.\",\n",
    "    \"The plot was okay, but the acting was amazing.\",\n",
    "    \"Terrible movie, I hated every minute of it.\"\n",
    "]\n",
    "test_model(model, test_texts, word_index, device=device)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "743ea84e-4525-4dec-bc0a-9098e10470d9",
   "metadata": {},
   "source": [
    "### 双向GRU的实际使用案例\n",
    "\n",
    "将上面的示例修改为使用双向GRU实现。\n",
    "\n",
    "**关键修改点解析**\n",
    "1. nn.GRU 初始化：\n",
    "- 在 IMDBGRU 类的 __init__ 方法中，为 nn.GRU 添加 bidirectional=True 参数。这个参数告诉 PyTorch 构建一个双向的 GRU 层。\n",
    "\n",
    "2. 全连接层输入维度：\n",
    "- 单向GRU的隐藏状态维度是hidden_size。\n",
    "- 双向GRU会将前向和后向的隐藏状态拼接起来，因此其隐藏状态维度会变为 2 * hidden_size。\n",
    "- 因此，需要将全连接层self.fc的输入维度从hidden_size调整为hidden_size * 2。\n",
    "\n",
    "3. forward 方法：\n",
    "\n",
    "- 在forward方法中，双向GRU的最后一个隐藏状态h_n 的形状是 [num_layers * num_directions, batch_size, hidden_size]。\n",
    "- 对于num_layers=1的情况，h_n的形状为 [2, batch_size, hidden_size]。其中：\n",
    "  - h_n[0, :, :] 是前向（从左到右）序列的最后一个隐藏状态。\n",
    "  - h_n[1, :, :] 是后向（从右到左）序列的最后一个隐藏状态。\n",
    "- 我们需要将这两个方向的隐藏状态沿着维度 1 拼接起来，形成一个形状为 [batch_size, hidden_size * 2] 的张量，再将其作为全连接层的输入。\n",
    "- 代码中通过 torch.cat((h_n[-2, :, :], h_n[-1, :, :]), dim=1) 实现了这个拼接操作。由于 num_layers=1 且 bidirectional=True，最后一个和倒数第二个隐藏状态就是两个方向的隐藏状态。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "363930f4-262a-4363-8931-4070d75c668a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from tensorflow.keras.datasets import imdb\n",
    "import numpy as np\n",
    "import re\n",
    "import string\n",
    "\n",
    "# 设置随机种子以确保可重复性\n",
    "torch.manual_seed(42)\n",
    "np.random.seed(42)\n",
    "\n",
    "# 1. 数据集准备\n",
    "class IMDBDataset(Dataset):\n",
    "    def __init__(self, data, labels, max_len=500):\n",
    "        self.data = data\n",
    "        self.labels = labels\n",
    "        self.max_len = max_len\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.data)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        # 截断或填充序列到 max_len\n",
    "        seq = self.data[idx][:self.max_len]\n",
    "        seq = np.pad(seq, (0, self.max_len - len(seq)), mode='constant') if len(seq) < self.max_len else seq\n",
    "        return torch.tensor(seq, dtype=torch.long), torch.tensor(self.labels[idx], dtype=torch.float)\n",
    "\n",
    "# 加载IMDB数据集和词典\n",
    "vocab_size = 20000  # 限制词汇表大小\n",
    "max_len = 500       # 最大序列长度\n",
    "(x_train, y_train), (x_test, y_test) = imdb.load_data(num_words=vocab_size)\n",
    "word_index = imdb.get_word_index()  # 获取单词到索引的映射\n",
    "\n",
    "# 创建PyTorch数据集\n",
    "train_dataset = IMDBDataset(x_train, y_train, max_len)\n",
    "test_dataset = IMDBDataset(x_test, y_test, max_len)\n",
    "\n",
    "# 创建数据加载器\n",
    "batch_size = 32\n",
    "train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)\n",
    "test_loader = DataLoader(test_dataset, batch_size=batch_size)\n",
    "\n",
    "# 2. 定义双向GRU模型\n",
    "# 修改点：\n",
    "# - `bidirectional=True` 参数\n",
    "# - 调整全连接层的输入维度\n",
    "# - 调整 `forward` 方法中隐藏状态的维度处理\n",
    "class BidirectionalIMDBGRU(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size, output_size, num_layers=1):\n",
    "        super(BidirectionalIMDBGRU, self).__init__()\n",
    "        self.hidden_size = hidden_size\n",
    "        self.num_layers = num_layers\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.gru = nn.GRU(\n",
    "            input_size=embed_size,\n",
    "            hidden_size=hidden_size,\n",
    "            num_layers=num_layers,\n",
    "            batch_first=True,\n",
    "            bidirectional=True  # <-- 修改点 1: 设置为双向\n",
    "        )\n",
    "        # 双向 GRU 的隐藏状态维度是 2 * hidden_size\n",
    "        self.fc = nn.Linear(hidden_size * 2, output_size) # <-- 修改点 2: 调整全连接层输入维度\n",
    "        self.sigmoid = nn.Sigmoid()\n",
    "\n",
    "    def forward(self, x):\n",
    "        # x: [batch_size, seq_len]\n",
    "        x = self.embedding(x) # [batch_size, seq_len, embed_size]\n",
    "        \n",
    "        # h_0 的形状为 (num_layers * num_directions, batch, hidden_size)\n",
    "        # 双向时 num_directions = 2\n",
    "        # 注意：这里我们不再需要手动传入 h_0，因为 GRU 在未提供时会自动初始化为零。\n",
    "        \n",
    "        # output 的形状为 [batch_size, seq_len, num_directions * hidden_size]\n",
    "        # h_n 的形状为 [num_layers * num_directions, batch_size, hidden_size]\n",
    "        output, h_n = self.gru(x)\n",
    "        \n",
    "        # 双向 GRU 的 h_n 包含了前向和后向的最后一个隐藏状态\n",
    "        # h_n 的第一层的前向隐藏状态是 h_n[0, :, :]\n",
    "        # h_n 的第一层的后向隐藏状态是 h_n[1, :, :]\n",
    "        # 我们可以将它们拼接起来作为分类器的输入\n",
    "        \n",
    "        # 提取最后一个时间步的前向和后向隐藏状态\n",
    "        h_n_forward = h_n[-2, :, :]\n",
    "        h_n_backward = h_n[-1, :, :]\n",
    "        \n",
    "        # 将两个隐藏状态在最后一个维度拼接\n",
    "        # out 的形状为 [batch_size, hidden_size * 2]\n",
    "        out = torch.cat((h_n_forward, h_n_backward), dim=1) # <-- 修改点 3: 拼接双向的隐藏状态\n",
    "        \n",
    "        out = self.fc(out)\n",
    "        out = self.sigmoid(out)\n",
    "        return out\n",
    "\n",
    "# 3. 模型参数\n",
    "embed_size = 128    # 词嵌入维度\n",
    "hidden_size = 256   # 隐藏状态维度\n",
    "output_size = 1     # 输出维度（二分类）\n",
    "num_layers = 1      # GRU层数\n",
    "\n",
    "# 初始化模型、损失函数和优化器\n",
    "model = BidirectionalIMDBGRU(vocab_size, embed_size, hidden_size, output_size, num_layers) # <-- 修改点\n",
    "criterion = nn.BCELoss()  # 二分类交叉熵损失\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 4. 训练模型\n",
    "def train_model(model, train_loader, criterion, optimizer, epochs=10, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for data, labels in train_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(data)\n",
    "            loss = criterion(output, labels)\n",
    "            loss.backward()\n",
    "            # 梯度裁剪防止爆炸\n",
    "            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1)\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f\"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}\")\n",
    "\n",
    "# 5. 评估模型\n",
    "def evaluate_model(model, test_loader, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for data, labels in test_loader:\n",
    "            data, labels = data.to(device), labels.to(device).view(-1, 1)\n",
    "            output = model(data)\n",
    "            predictions = (output >= 0.5).float()\n",
    "            correct += (predictions == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    accuracy = correct / total * 100\n",
    "    print(f\"Test Accuracy: {accuracy:.2f}%\")\n",
    "\n",
    "# 6. 测试功能：处理示例评论\n",
    "def text_to_sequence(text, word_index, max_len=500):\n",
    "    # 文本预处理：小写、去除标点\n",
    "    text = text.lower()\n",
    "    text = re.sub(f'[{string.punctuation}]', '', text)\n",
    "    # 分词\n",
    "    tokens = text.split()\n",
    "    # 转换为单词索引\n",
    "    sequence = [word_index.get(word, 0) for word in tokens if word_index.get(word, 0) < vocab_size]\n",
    "    # 截断或填充\n",
    "    sequence = sequence[:max_len]\n",
    "    sequence = np.pad(sequence, (0, max_len - len(sequence)), mode='constant') if len(sequence) < max_len else sequence\n",
    "    return torch.tensor([sequence], dtype=torch.long)  # [1, max_len]\n",
    "\n",
    "def test_model(model, texts, word_index, device='cpu'):\n",
    "    model = model.to(device)\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        for text in texts:\n",
    "            # 转换为序列\n",
    "            seq = text_to_sequence(text, word_index, max_len)\n",
    "            seq = seq.to(device)\n",
    "            # 预测\n",
    "            output = model(seq)  # [1, 1]\n",
    "            prob = output.item()\n",
    "            sentiment = \"Positive\" if prob >= 0.5 else \"Negative\"\n",
    "            print(f\"Text: {text}\")\n",
    "            print(f\"Sentiment: {sentiment}, Probability: {prob:.4f}\\n\")\n",
    "\n",
    "# 7. 运行训练、评估和测试\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "train_model(model, train_loader, criterion, optimizer, epochs=10, device=device)\n",
    "evaluate_model(model, test_loader, device=device)\n",
    "\n",
    "# 示例评论\n",
    "test_texts = [\n",
    "    \"I absolutely love this movie, it's fantastic and thrilling!\",\n",
    "    \"This film was boring and a complete waste of time.\",\n",
    "    \"The plot was okay, but the acting was amazing.\",\n",
    "    \"Terrible movie, I hated every minute of it.\"\n",
    "]\n",
    "test_model(model, test_texts, word_index, device=device)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7fd2444f-5248-4c1d-b1b2-6d4e164bb14e",
   "metadata": {},
   "source": [
    "## Seq2Seq示例"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba5f6210-d5cf-4a98-803c-148ad2878cf4",
   "metadata": {},
   "source": [
    "### 简单的seq2seq模型\n",
    "\n",
    "下面是一个使用PyTorch构建Encoder-Decoder模型的简单示例。为了方便演示，我们省略了数据预处理部分，只展示模型的核心结构。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cf8534bd-3cd2-40fc-ba18-9e3ccafdecb9",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "\n",
    "# 定义编码器\n",
    "class Encoder(nn.Module):\n",
    "    def __init__(self, input_dim, embed_dim, hidden_dim, n_layers):\n",
    "        super().__init__()\n",
    "        \n",
    "        self.hidden_dim = hidden_dim\n",
    "        self.n_layers = n_layers\n",
    "        \n",
    "        # 嵌入层：将输入的词语索引转换为词向量\n",
    "        self.embedding = nn.Embedding(input_dim, embed_dim)\n",
    "        \n",
    "        # LSTM层：处理序列数据\n",
    "        self.lstm = nn.LSTM(embed_dim, hidden_dim, n_layers)\n",
    "\n",
    "    def forward(self, src):\n",
    "        # src: [sequence_length, batch_size]\n",
    "        \n",
    "        # 词嵌入：[sequence_length, batch_size, embed_dim]\n",
    "        embedded = self.embedding(src)\n",
    "        \n",
    "        # LSTM前向传播\n",
    "        # outputs: [sequence_length, batch_size, hidden_dim * num_directions]\n",
    "        # (hidden, cell)是最终的隐藏状态和细胞状态，用于传递给解码器\n",
    "        outputs, (hidden, cell) = self.lstm(embedded)\n",
    "        \n",
    "        # 返回最终的隐藏状态和细胞状态\n",
    "        return hidden, cell\n",
    "\n",
    "# 定义解码器\n",
    "class Decoder(nn.Module):\n",
    "    def __init__(self, output_dim, embed_dim, hidden_dim, n_layers):\n",
    "        super().__init__()\n",
    "        \n",
    "        self.output_dim = output_dim\n",
    "        self.hidden_dim = hidden_dim\n",
    "        self.n_layers = n_layers\n",
    "        \n",
    "        # 嵌入层\n",
    "        self.embedding = nn.Embedding(output_dim, embed_dim)\n",
    "        \n",
    "        # LSTM层\n",
    "        self.lstm = nn.LSTM(embed_dim, hidden_dim, n_layers)\n",
    "        \n",
    "        # 线性层：将LSTM的输出映射到词汇表大小\n",
    "        self.fc_out = nn.Linear(hidden_dim, output_dim)\n",
    "\n",
    "    def forward(self, input, hidden, cell):\n",
    "        # input: [batch_size] -> 解码器的输入通常是一个词，所以需要增加一个维度\n",
    "        input = input.unsqueeze(0)\n",
    "        \n",
    "        # 嵌入\n",
    "        embedded = self.embedding(input)\n",
    "        \n",
    "        # LSTM前向传播，初始隐藏状态和细胞状态来自编码器\n",
    "        output, (hidden, cell) = self.lstm(embedded, (hidden, cell))\n",
    "        \n",
    "        # output: [1, batch_size, hidden_dim]\n",
    "        # 将 output 展平，以便输入到全连接层\n",
    "        prediction = self.fc_out(output.squeeze(0))\n",
    "        \n",
    "        return prediction, hidden, cell\n",
    "\n",
    "# 定义Seq2Seq模型\n",
    "class Seq2Seq(nn.Module):\n",
    "    def __init__(self, encoder, decoder):\n",
    "        super().__init__()\n",
    "        \n",
    "        self.encoder = encoder\n",
    "        self.decoder = decoder\n",
    "        \n",
    "        # 确保编码器和解码器的隐藏层维度和层数匹配\n",
    "        assert encoder.hidden_dim == decoder.hidden_dim, \\\n",
    "            \"Hidden dimensions of encoder and decoder must be equal!\"\n",
    "        assert encoder.n_layers == decoder.n_layers, \\\n",
    "            \"Number of layers of encoder and decoder must be equal!\"\n",
    "\n",
    "    def forward(self, src, trg, teacher_forcing_ratio=0.5):\n",
    "        # src: [src_len, batch_size]\n",
    "        # trg: [trg_len, batch_size]\n",
    "        \n",
    "        # 获得目标序列的长度\n",
    "        trg_len = trg.shape[0]\n",
    "        batch_size = trg.shape[1]\n",
    "        \n",
    "        # 存储解码器输出的张量\n",
    "        outputs = torch.zeros(trg_len, batch_size, self.decoder.output_dim)\n",
    "        \n",
    "        # 编码器获得隐藏状态和细胞状态\n",
    "        hidden, cell = self.encoder(src)\n",
    "        \n",
    "        # 解码器的第一个输入是起始符 <sos>\n",
    "        input = trg[0, :]\n",
    "        \n",
    "        for t in range(1, trg_len):\n",
    "            # 将前一步的隐藏状态、细胞状态和当前输入传给解码器\n",
    "            output, hidden, cell = self.decoder(input, hidden, cell)\n",
    "            \n",
    "            # 将输出保存到 outputs 中\n",
    "            outputs[t] = output\n",
    "            \n",
    "            # 决定是否使用“教师强制”（Teacher Forcing）\n",
    "            # 教师强制：用真实的目标序列词作为下一步输入\n",
    "            # 否则：用模型预测出的最可能的词作为下一步输入\n",
    "            teacher_force = random.random() < teacher_forcing_ratio\n",
    "            top1 = output.argmax(1)\n",
    "            \n",
    "            input = trg[t] if teacher_force else top1\n",
    "        \n",
    "        return outputs\n",
    "\n",
    "# 实例化模型\n",
    "INPUT_DIM = 5000  # 源语言词汇量\n",
    "OUTPUT_DIM = 5000 # 目标语言词汇量\n",
    "EMBED_DIM = 256\n",
    "HIDDEN_DIM = 512\n",
    "N_LAYERS = 2\n",
    "\n",
    "encoder = Encoder(INPUT_DIM, EMBED_DIM, HIDDEN_DIM, N_LAYERS)\n",
    "decoder = Decoder(OUTPUT_DIM, EMBED_DIM, HIDDEN_DIM, N_LAYERS)\n",
    "model = Seq2Seq(encoder, decoder)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15706394-f5bf-4b66-bb9b-3969237fac8a",
   "metadata": {},
   "source": [
    "### 机器翻译模型示例"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "863a9f76-5ac1-43e2-ac73-496a123ca0e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import random\n",
    "from collections import Counter\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "\n",
    "# 数据集与词表\n",
    "pairs = [\n",
    "    (\"你好\", \"Hello\"),\n",
    "    (\"谢谢\", \"Thanks\"),\n",
    "    (\"我爱你\", \"I love you\"),\n",
    "    (\"早上好\", \"Good morning\"),\n",
    "    (\"晚安\", \"Good night\"),\n",
    "    ('天气很好', 'The weather is good'),\n",
    "    ('我喜欢学习编程', 'I like to learn programming'),\n",
    "    ('他正在看电影', 'He is watching a movie'),\n",
    "    ('请告诉我你的名字', 'Please tell me your name'),\n",
    "    ('她是一个学生', 'She is a student'),\n",
    "    ('我们是朋友', 'We are friends'),\n",
    "    ('这本书很有趣', 'This book is interesting'),\n",
    "]\n",
    "\n",
    "# 特殊标记\n",
    "SOS_token = \"<SOS>\"\n",
    "EOS_token = \"<EOS>\"\n",
    "PAD_token = \"<PAD>\"\n",
    "UNK_token = \"<UNK>\"  # 添加未知词标记以处理未见词\n",
    "\n",
    "# 改进的词汇表构建函数，确保包含所有单词\n",
    "def build_vocab(sentences, is_target=False):\n",
    "    counter = Counter()\n",
    "    if is_target:\n",
    "        # 对于英文，按单词分割\n",
    "        for sent in sentences:\n",
    "            words = sent.split()  # 直接按空格分割\n",
    "            counter.update(words)\n",
    "    else:\n",
    "        # 对于中文，按字符分割\n",
    "        for sent in sentences:\n",
    "            counter.update(list(sent))\n",
    "    # 添加特殊标记\n",
    "    vocab = {PAD_token: 0, SOS_token: 1, EOS_token: 2, UNK_token: 3}\n",
    "    vocab.update({word: i + 4 for i, word in enumerate(counter.keys())})\n",
    "    return vocab, {i: word for word, i in vocab.items()}\n",
    "\n",
    "# 分离源语言和目标语言句子\n",
    "src_sentences, tgt_sentences = zip(*pairs)\n",
    "src_vocab, src_itos = build_vocab(src_sentences, is_target=False)\n",
    "tgt_vocab, tgt_itos = build_vocab(tgt_sentences, is_target=True)\n",
    "\n",
    "# 数据集类\n",
    "class TranslationDataset(Dataset):\n",
    "    def __init__(self, pairs, src_vocab, tgt_vocab):\n",
    "        self.pairs = pairs\n",
    "        self.src_vocab = src_vocab\n",
    "        self.tgt_vocab = tgt_vocab\n",
    "    \n",
    "    def __len__(self):\n",
    "        return len(self.pairs)\n",
    "    \n",
    "    def __getitem__(self, idx):\n",
    "        src, tgt = self.pairs[idx]\n",
    "        # 中文按字符分割\n",
    "        src_indices = [self.src_vocab[SOS_token]] + [self.src_vocab.get(c, self.src_vocab[UNK_token]) for c in src] + [self.src_vocab[EOS_token]]\n",
    "        # 英文按单词分割\n",
    "        tgt_words = tgt.split()\n",
    "        tgt_indices = [self.tgt_vocab[SOS_token]] + [self.tgt_vocab.get(w, self.tgt_vocab[UNK_token]) for w in tgt_words] + [self.tgt_vocab[EOS_token]]\n",
    "        return torch.tensor(src_indices), torch.tensor(tgt_indices)\n",
    "\n",
    "# 模型定义\n",
    "class Encoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Encoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size, hidden_size, batch_first=True)\n",
    "    \n",
    "    def forward(self, src):\n",
    "        embedded = self.embedding(src)\n",
    "        outputs, (hidden, cell) = self.lstm(embedded)\n",
    "        return outputs, hidden, cell\n",
    "\n",
    "class BahdanauAttention(nn.Module):\n",
    "    def __init__(self, hidden_size):\n",
    "        super(BahdanauAttention, self).__init__()\n",
    "        self.Wa = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Ua = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Va = nn.Linear(hidden_size, 1)\n",
    "    \n",
    "    def forward(self, decoder_hidden, encoder_outputs):\n",
    "        score = self.Va(torch.tanh(self.Wa(decoder_hidden.unsqueeze(1)) + self.Ua(encoder_outputs)))\n",
    "        attn_weights = torch.softmax(score, dim=1)\n",
    "        context = torch.bmm(attn_weights.transpose(1, 2), encoder_outputs)\n",
    "        return context.squeeze(1), attn_weights.squeeze(2)\n",
    "\n",
    "class Decoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Decoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size + hidden_size, hidden_size, batch_first=True)\n",
    "        self.attention = BahdanauAttention(hidden_size)\n",
    "        self.out = nn.Linear(hidden_size, vocab_size)\n",
    "    \n",
    "    def forward(self, tgt, hidden, cell, encoder_outputs):\n",
    "        embedded = self.embedding(tgt)\n",
    "        context, attn_weights = self.attention(hidden[-1], encoder_outputs)\n",
    "        lstm_input = torch.cat((embedded, context.unsqueeze(1)), dim=2)\n",
    "        output, (hidden, cell) = self.lstm(lstm_input, (hidden, cell))\n",
    "        output = self.out(output.squeeze(1))\n",
    "        return output, hidden, cell, attn_weights\n",
    "\n",
    "class Seq2Seq(nn.Module):\n",
    "    def __init__(self, encoder, decoder):\n",
    "        super(Seq2Seq, self).__init__()\n",
    "        self.encoder = encoder\n",
    "        self.decoder = decoder\n",
    "    \n",
    "    def forward(self, src, tgt, teacher_forcing_ratio=0.5):\n",
    "        batch_size = src.size(0)\n",
    "        tgt_len = tgt.size(1)\n",
    "        tgt_vocab_size = self.decoder.out.out_features\n",
    "        \n",
    "        outputs = torch.zeros(batch_size, tgt_len, tgt_vocab_size).to(src.device)\n",
    "        encoder_outputs, hidden, cell = self.encoder(src)\n",
    "        \n",
    "        input = tgt[:, 0].unsqueeze(1)  # <SOS>\n",
    "        for t in range(1, tgt_len):\n",
    "            output, hidden, cell, _ = self.decoder(input, hidden, cell, encoder_outputs)\n",
    "            outputs[:, t, :] = output\n",
    "            teacher_force = random.random() < teacher_forcing_ratio\n",
    "            input = tgt[:, t].unsqueeze(1) if teacher_force else output.argmax(1).unsqueeze(1)\n",
    "        \n",
    "        return outputs\n",
    "\n",
    "# 数据准备\n",
    "dataset = TranslationDataset(pairs, src_vocab, tgt_vocab)\n",
    "dataloader = DataLoader(dataset, batch_size=2, shuffle=True, collate_fn=lambda x: (\n",
    "    nn.utils.rnn.pad_sequence([item[0] for item in x], batch_first=True, padding_value=src_vocab[PAD_token]),\n",
    "    nn.utils.rnn.pad_sequence([item[1] for item in x], batch_first=True, padding_value=tgt_vocab[PAD_token])\n",
    "))\n",
    "\n",
    "# 模型参数\n",
    "embed_size = 128\n",
    "hidden_size = 256\n",
    "src_vocab_size = len(src_vocab)\n",
    "tgt_vocab_size = len(tgt_vocab)\n",
    "\n",
    "encoder = Encoder(src_vocab_size, embed_size, hidden_size)\n",
    "decoder = Decoder(tgt_vocab_size, embed_size, hidden_size)\n",
    "model = Seq2Seq(encoder, decoder)\n",
    "\n",
    "# 训练设置\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "model = model.to(device)\n",
    "criterion = nn.CrossEntropyLoss(ignore_index=tgt_vocab[PAD_token])\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 训练循环\n",
    "def train(model, dataloader, criterion, optimizer, epochs=20):\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for src, tgt in dataloader:\n",
    "            src, tgt = src.to(device), tgt.to(device)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(src, tgt)\n",
    "            loss = criterion(output[:, 1:, :].reshape(-1, tgt_vocab_size), tgt[:, 1:].reshape(-1))\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f'Epoch {epoch+1}, Loss: {total_loss / len(dataloader)}')\n",
    "\n",
    "# 推理函数\n",
    "def translate(model, sentence, src_vocab, tgt_vocab, src_itos, tgt_itos, max_len=20):\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        src_indices = [src_vocab[SOS_token]] + [src_vocab.get(c, src_vocab[UNK_token]) for c in sentence] + [src_vocab[EOS_token]]\n",
    "        src_tensor = torch.tensor([src_indices], dtype=torch.long).to(device)\n",
    "        \n",
    "        encoder_outputs, hidden, cell = model.encoder(src_tensor)\n",
    "        input = torch.tensor([[tgt_vocab[SOS_token]]]).to(device)\n",
    "        output_words = []\n",
    "        \n",
    "        for _ in range(max_len):\n",
    "            output, hidden, cell, _ = model.decoder(input, hidden, cell, encoder_outputs)\n",
    "            pred_token = output.argmax(1).item()\n",
    "            if pred_token == tgt_vocab[EOS_token]:\n",
    "                break\n",
    "            output_words.append(tgt_itos[pred_token])\n",
    "            input = torch.tensor([[pred_token]]).to(device)\n",
    "        \n",
    "        return ' '.join(output_words)\n",
    "\n",
    "# 训练模型\n",
    "train(model, dataloader, criterion, optimizer)\n",
    "\n",
    "# 测试翻译\n",
    "print(\"\\n===== 测试翻译 =====\")\n",
    "test_sentences = [\"你好\", \"我爱你\", \"这本书很有趣\", \"请告诉我你的名字\" ]\n",
    "for sent in test_sentences:\n",
    "    translation = translate(model, sent, src_vocab, tgt_vocab, src_itos, tgt_itos)\n",
    "    print(f\"Chinese: {sent} -> English: {translation}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a61a6674-f247-45ac-bbe2-8494c0400b95",
   "metadata": {},
   "source": [
    "### 示例2：SRE告警处理建议\n",
    "\n",
    "**代码说明**\n",
    "- 数据集：使用SRE相关的告警-操作对，模拟真实场景。告警消息和操作建议都按单词分割。\n",
    "- 模型：与翻译示例相同，使用LSTM+Bahdanau注意力，适合SRE的序列映射任务。\n",
    "- 预处理：词汇表按单词构建，添加<UNK>处理未见词。\n",
    "- 训练：小数据集，20个epoch，Adam优化器，Teacher Forcing比率0.5。\n",
    "- 推理：handle_alert函数将告警消息翻译为操作建议。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c0f2158e-a994-49ce-be5b-b0b526b5e0bc",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import random\n",
    "from collections import Counter\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "\n",
    "# 数据集\n",
    "pairs = [\n",
    "    (\"CPU usage > 90%\", \"Restart service\"),\n",
    "    (\"Memory leak detected\", \"Increase memory allocation\"),\n",
    "    (\"Disk space low\", \"Clean up disk\"),\n",
    "    (\"Service A timeout\", \"Check network connection\"),\n",
    "    (\"High latency in API\", \"Scale up servers\"),\n",
    "    (\"Database connection failed\", \"Restart database\"),\n",
    "]\n",
    "\n",
    "# 特殊标记\n",
    "SOS_token = \"<SOS>\"\n",
    "EOS_token = \"<EOS>\"\n",
    "PAD_token = \"<PAD>\"\n",
    "UNK_token = \"<UNK>\"\n",
    "\n",
    "# 构建词表\n",
    "def build_vocab(sentences):\n",
    "    counter = Counter()\n",
    "    for sent in sentences:\n",
    "        words = sent.split()  # 按单词分割\n",
    "        counter.update(words)\n",
    "    vocab = {PAD_token: 0, SOS_token: 1, EOS_token: 2, UNK_token: 3}\n",
    "    vocab.update({word: i + 4 for i, word in enumerate(counter.keys())})\n",
    "    return vocab, {i: word for word, i in vocab.items()}\n",
    "\n",
    "src_sentences, tgt_sentences = zip(*pairs)\n",
    "src_vocab, src_itos = build_vocab(src_sentences)\n",
    "tgt_vocab, tgt_itos = build_vocab(tgt_sentences)\n",
    "\n",
    "# 数据集类\n",
    "class AlertDataset(Dataset):\n",
    "    def __init__(self, pairs, src_vocab, tgt_vocab):\n",
    "        self.pairs = pairs\n",
    "        self.src_vocab = src_vocab\n",
    "        self.tgt_vocab = tgt_vocab\n",
    "    \n",
    "    def __len__(self):\n",
    "        return len(self.pairs)\n",
    "    \n",
    "    def __getitem__(self, idx):\n",
    "        src, tgt = self.pairs[idx]\n",
    "        src_indices = [self.src_vocab[SOS_token]] + [self.src_vocab.get(w, self.src_vocab[UNK_token]) for w in src.split()] + [self.src_vocab[EOS_token]]\n",
    "        tgt_indices = [self.tgt_vocab[SOS_token]] + [self.tgt_vocab.get(w, self.tgt_vocab[UNK_token]) for w in tgt.split()] + [self.tgt_vocab[EOS_token]]\n",
    "        return torch.tensor(src_indices), torch.tensor(tgt_indices)\n",
    "\n",
    "# 模型定义（与翻译示例相同）\n",
    "class Encoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Encoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size, hidden_size, batch_first=True)\n",
    "    \n",
    "    def forward(self, src):\n",
    "        embedded = self.embedding(src)\n",
    "        outputs, (hidden, cell) = self.lstm(embedded)\n",
    "        return outputs, hidden, cell\n",
    "\n",
    "class BahdanauAttention(nn.Module):\n",
    "    def __init__(self, hidden_size):\n",
    "        super(BahdanauAttention, self).__init__()\n",
    "        self.Wa = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Ua = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Va = nn.Linear(hidden_size, 1)\n",
    "    \n",
    "    def forward(self, decoder_hidden, encoder_outputs):\n",
    "        score = self.Va(torch.tanh(self.Wa(decoder_hidden.unsqueeze(1)) + self.Ua(encoder_outputs)))\n",
    "        attn_weights = torch.softmax(score, dim=1)\n",
    "        context = torch.bmm(attn_weights.transpose(1, 2), encoder_outputs)\n",
    "        return context.squeeze(1), attn_weights.squeeze(2)\n",
    "\n",
    "class Decoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Decoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size + hidden_size, hidden_size, batch_first=True)\n",
    "        self.attention = BahdanauAttention(hidden_size)\n",
    "        self.out = nn.Linear(hidden_size, vocab_size)\n",
    "    \n",
    "    def forward(self, tgt, hidden, cell, encoder_outputs):\n",
    "        embedded = self.embedding(tgt)\n",
    "        context, attn_weights = self.attention(hidden[-1], encoder_outputs)\n",
    "        lstm_input = torch.cat((embedded, context.unsqueeze(1)), dim=2)\n",
    "        output, (hidden, cell) = self.lstm(lstm_input, (hidden, cell))\n",
    "        output = self.out(output.squeeze(1))\n",
    "        return output, hidden, cell, attn_weights\n",
    "\n",
    "class Seq2Seq(nn.Module):\n",
    "    def __init__(self, encoder, decoder):\n",
    "        super(Seq2Seq, self).__init__()\n",
    "        self.encoder = encoder\n",
    "        self.decoder = decoder\n",
    "    \n",
    "    def forward(self, src, tgt, teacher_forcing_ratio=0.5):\n",
    "        batch_size = src.size(0)\n",
    "        tgt_len = tgt.size(1)\n",
    "        tgt_vocab_size = self.decoder.out.out_features\n",
    "        \n",
    "        outputs = torch.zeros(batch_size, tgt_len, tgt_vocab_size).to(src.device)\n",
    "        encoder_outputs, hidden, cell = self.encoder(src)\n",
    "        \n",
    "        input = tgt[:, 0].unsqueeze(1)  # <SOS>\n",
    "        for t in range(1, tgt_len):\n",
    "            output, hidden, cell, _ = self.decoder(input, hidden, cell, encoder_outputs)\n",
    "            outputs[:, t, :] = output\n",
    "            teacher_force = random.random() < teacher_forcing_ratio\n",
    "            input = tgt[:, t].unsqueeze(1) if teacher_force else output.argmax(1).unsqueeze(1)\n",
    "        \n",
    "        return outputs\n",
    "\n",
    "# 数据准备\n",
    "dataset = AlertDataset(pairs, src_vocab, tgt_vocab)\n",
    "dataloader = DataLoader(dataset, batch_size=2, shuffle=True, collate_fn=lambda x: (\n",
    "    nn.utils.rnn.pad_sequence([item[0] for item in x], batch_first=True, padding_value=src_vocab[PAD_token]),\n",
    "    nn.utils.rnn.pad_sequence([item[1] for item in x], batch_first=True, padding_value=tgt_vocab[PAD_token])\n",
    "))\n",
    "\n",
    "# 模型参数\n",
    "embed_size = 128\n",
    "hidden_size = 256\n",
    "src_vocab_size = len(src_vocab)\n",
    "tgt_vocab_size = len(tgt_vocab)\n",
    "\n",
    "encoder = Encoder(src_vocab_size, embed_size, hidden_size)\n",
    "decoder = Decoder(tgt_vocab_size, embed_size, hidden_size)\n",
    "model = Seq2Seq(encoder, decoder)\n",
    "\n",
    "# 训练设置\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "model = model.to(device)\n",
    "criterion = nn.CrossEntropyLoss(ignore_index=tgt_vocab[PAD_token])\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 训练循环\n",
    "def train(model, dataloader, criterion, optimizer, epochs=20):\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for src, tgt in dataloader:\n",
    "            src, tgt = src.to(device), tgt.to(device)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(src, tgt)\n",
    "            loss = criterion(output[:, 1:, :].reshape(-1, tgt_vocab_size), tgt[:, 1:].reshape(-1))\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f'Epoch {epoch+1}, Loss: {total_loss / len(dataloader)}')\n",
    "\n",
    "# 推理函数\n",
    "def handle_alert(model, alert, src_vocab, tgt_vocab, src_itos, tgt_itos, max_len=20):\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        src_indices = [src_vocab[SOS_token]] + [src_vocab.get(w, src_vocab[UNK_token]) for w in alert.split()] + [src_vocab[EOS_token]]\n",
    "        src_tensor = torch.tensor([src_indices], dtype=torch.long).to(device)\n",
    "        \n",
    "        encoder_outputs, hidden, cell = model.encoder(src_tensor)\n",
    "        input = torch.tensor([[tgt_vocab[SOS_token]]]).to(device)\n",
    "        output_words = []\n",
    "        \n",
    "        for _ in range(max_len):\n",
    "            output, hidden, cell, _ = model.decoder(input, hidden, cell, encoder_outputs)\n",
    "            pred_token = output.argmax(1).item()\n",
    "            if pred_token == tgt_vocab[EOS_token]:\n",
    "                break\n",
    "            output_words.append(tgt_itos[pred_token])\n",
    "            input = torch.tensor([[pred_token]]).to(device)\n",
    "        \n",
    "        return ' '.join(output_words)\n",
    "\n",
    "# 训练模型\n",
    "train(model, dataloader, criterion, optimizer)\n",
    "\n",
    "# 测试告警处理\n",
    "print(\"\\n===== 测试告警处理 =====\")\n",
    "test_alerts = [\"CPU usa > 90% xxx xxx xxx\", \"Disk space low xxx\", \"High latency in API\"]\n",
    "for alert in test_alerts:\n",
    "    suggestion = handle_alert(model, alert, src_vocab, tgt_vocab, src_itos, tgt_itos)\n",
    "    print(f\"Alert: {alert} -> Suggestion: {suggestion}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf8937b0-fd9a-49bf-8ad2-ce79b769d5af",
   "metadata": {},
   "source": [
    "### 示例3：日志分析与异常检测"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8d074d1a-68b3-41f0-8b3b-06179d4b479b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import random\n",
    "from collections import Counter\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "\n",
    "# 模拟日志数据集：(日志序列, 标签序列)\n",
    "# 日志序列：按空格分割的日志消息列表（模拟多个日志条目）\n",
    "# 标签序列：每个日志的标签 (\"normal\" 或 \"anomalous\")\n",
    "pairs = [\n",
    "    (\"[INFO] System started successfully\", \"normal\"),\n",
    "    (\"[ERROR] Connection timeout occurred\", \"anomalous\"),\n",
    "    (\"[WARN] Disk space low at 90%\", \"anomalous\"),\n",
    "    (\"[INFO] User logged in\", \"normal\"),\n",
    "    (\"[FATAL] Database crash detected\", \"anomalous\"),\n",
    "    (\"[DEBUG] Request processed in 10ms\", \"normal\"),\n",
    "    (\"[ERROR] Invalid credentials provided\", \"anomalous\"),\n",
    "    (\"[INFO] Backup completed\", \"normal\"),\n",
    "    (\"[WARN] High CPU usage 95%\", \"anomalous\"),\n",
    "    (\"[INFO] Service restarted\", \"normal\"),\n",
    "    (\"[ERROR] Network failure\", \"anomalous\"),\n",
    "    (\"[INFO] All systems operational\", \"normal\"),\n",
    "]\n",
    "\n",
    "# 特殊标记\n",
    "SOS_token = \"<SOS>\"\n",
    "EOS_token = \"<EOS>\"\n",
    "PAD_token = \"<PAD>\"\n",
    "UNK_token = \"<UNK>\"\n",
    "\n",
    "# 构建词表\n",
    "def build_vocab(sentences, is_label=False):\n",
    "    counter = Counter()\n",
    "    for sent in sentences:\n",
    "        words = sent.split()  # 按空格分割日志或标签\n",
    "        counter.update(words)\n",
    "    vocab = {PAD_token: 0, SOS_token: 1, EOS_token: 2, UNK_token: 3}\n",
    "    vocab.update({word: i + 4 for i, word in enumerate(counter.keys())})\n",
    "    return vocab, {i: word for word, i in vocab.items()}\n",
    "\n",
    "src_sentences, tgt_sentences = zip(*pairs)\n",
    "src_vocab, src_itos = build_vocab(src_sentences)  # 日志词表\n",
    "tgt_vocab, tgt_itos = build_vocab(tgt_sentences)  # 标签词表（简单标签如 \"normal\", \"anomalous\"）\n",
    "\n",
    "# 数据集类\n",
    "class LogDataset(Dataset):\n",
    "    def __init__(self, pairs, src_vocab, tgt_vocab):\n",
    "        self.pairs = pairs\n",
    "        self.src_vocab = src_vocab\n",
    "        self.tgt_vocab = tgt_vocab\n",
    "    \n",
    "    def __len__(self):\n",
    "        return len(self.pairs)\n",
    "    \n",
    "    def __getitem__(self, idx):\n",
    "        src, tgt = self.pairs[idx]\n",
    "        # 日志按单词分割\n",
    "        src_indices = [self.src_vocab[SOS_token]] + [self.src_vocab.get(w, self.src_vocab[UNK_token]) for w in src.split()] + [self.src_vocab[EOS_token]]\n",
    "        # 标签按单词分割（这里标签是单个词，但可扩展为序列）\n",
    "        tgt_indices = [self.tgt_vocab[SOS_token]] + [self.tgt_vocab.get(w, self.tgt_vocab[UNK_token]) for w in tgt.split()] + [self.tgt_vocab[EOS_token]]\n",
    "        return torch.tensor(src_indices), torch.tensor(tgt_indices)\n",
    "\n",
    "# 模型定义\n",
    "class Encoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Encoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size, hidden_size, batch_first=True)\n",
    "    \n",
    "    def forward(self, src):\n",
    "        embedded = self.embedding(src)\n",
    "        outputs, (hidden, cell) = self.lstm(embedded)\n",
    "        return outputs, hidden, cell\n",
    "\n",
    "class BahdanauAttention(nn.Module):\n",
    "    def __init__(self, hidden_size):\n",
    "        super(BahdanauAttention, self).__init__()\n",
    "        self.Wa = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Ua = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Va = nn.Linear(hidden_size, 1)\n",
    "    \n",
    "    def forward(self, decoder_hidden, encoder_outputs):\n",
    "        score = self.Va(torch.tanh(self.Wa(decoder_hidden.unsqueeze(1)) + self.Ua(encoder_outputs)))\n",
    "        attn_weights = torch.softmax(score, dim=1)\n",
    "        context = torch.bmm(attn_weights.transpose(1, 2), encoder_outputs)\n",
    "        return context.squeeze(1), attn_weights.squeeze(2)\n",
    "\n",
    "class Decoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Decoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size + hidden_size, hidden_size, batch_first=True)\n",
    "        self.attention = BahdanauAttention(hidden_size)\n",
    "        self.out = nn.Linear(hidden_size, vocab_size)\n",
    "    \n",
    "    def forward(self, tgt, hidden, cell, encoder_outputs):\n",
    "        embedded = self.embedding(tgt)\n",
    "        context, attn_weights = self.attention(hidden[-1], encoder_outputs)\n",
    "        lstm_input = torch.cat((embedded, context.unsqueeze(1)), dim=2)\n",
    "        output, (hidden, cell) = self.lstm(lstm_input, (hidden, cell))\n",
    "        output = self.out(output.squeeze(1))\n",
    "        return output, hidden, cell, attn_weights\n",
    "\n",
    "class Seq2Seq(nn.Module):\n",
    "    def __init__(self, encoder, decoder):\n",
    "        super(Seq2Seq, self).__init__()\n",
    "        self.encoder = encoder\n",
    "        self.decoder = decoder\n",
    "    \n",
    "    def forward(self, src, tgt, teacher_forcing_ratio=0.5):\n",
    "        batch_size = src.size(0)\n",
    "        tgt_len = tgt.size(1)\n",
    "        tgt_vocab_size = self.decoder.out.out_features\n",
    "        \n",
    "        outputs = torch.zeros(batch_size, tgt_len, tgt_vocab_size).to(src.device)\n",
    "        encoder_outputs, hidden, cell = self.encoder(src)\n",
    "        \n",
    "        input = tgt[:, 0].unsqueeze(1)  # <SOS>\n",
    "        for t in range(1, tgt_len):\n",
    "            output, hidden, cell, _ = self.decoder(input, hidden, cell, encoder_outputs)\n",
    "            outputs[:, t, :] = output\n",
    "            teacher_force = random.random() < teacher_forcing_ratio\n",
    "            input = tgt[:, t].unsqueeze(1) if teacher_force else output.argmax(1).unsqueeze(1)\n",
    "        \n",
    "        return outputs\n",
    "\n",
    "# 数据准备\n",
    "dataset = LogDataset(pairs, src_vocab, tgt_vocab)\n",
    "dataloader = DataLoader(dataset, batch_size=2, shuffle=True, collate_fn=lambda x: (\n",
    "    nn.utils.rnn.pad_sequence([item[0] for item in x], batch_first=True, padding_value=src_vocab[PAD_token]),\n",
    "    nn.utils.rnn.pad_sequence([item[1] for item in x], batch_first=True, padding_value=tgt_vocab[PAD_token])\n",
    "))\n",
    "\n",
    "# 模型参数\n",
    "embed_size = 128\n",
    "hidden_size = 256\n",
    "src_vocab_size = len(src_vocab)\n",
    "tgt_vocab_size = len(tgt_vocab)\n",
    "\n",
    "encoder = Encoder(src_vocab_size, embed_size, hidden_size)\n",
    "decoder = Decoder(tgt_vocab_size, embed_size, hidden_size)\n",
    "model = Seq2Seq(encoder, decoder)\n",
    "\n",
    "# 训练设置\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "model = model.to(device)\n",
    "criterion = nn.CrossEntropyLoss(ignore_index=tgt_vocab[PAD_token])\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 训练循环\n",
    "def train(model, dataloader, criterion, optimizer, epochs=20):\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for src, tgt in dataloader:\n",
    "            src, tgt = src.to(device), tgt.to(device)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(src, tgt)\n",
    "            loss = criterion(output[:, 1:, :].reshape(-1, tgt_vocab_size), tgt[:, 1:].reshape(-1))\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f'Epoch {epoch+1}, Loss: {total_loss / len(dataloader)}')\n",
    "\n",
    "# 异常检测函数（推理）\n",
    "def detect_anomaly(model, log_message, src_vocab, tgt_vocab, src_itos, tgt_itos, max_len=5):\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        src_indices = [src_vocab[SOS_token]] + [src_vocab.get(w, src_vocab[UNK_token]) for w in log_message.split()] + [src_vocab[EOS_token]]\n",
    "        src_tensor = torch.tensor([src_indices], dtype=torch.long).to(device)\n",
    "        \n",
    "        encoder_outputs, hidden, cell = model.encoder(src_tensor)\n",
    "        input = torch.tensor([[tgt_vocab[SOS_token]]]).to(device)\n",
    "        output_labels = []\n",
    "        \n",
    "        for _ in range(max_len):\n",
    "            output, hidden, cell, _ = model.decoder(input, hidden, cell, encoder_outputs)\n",
    "            pred_token = output.argmax(1).item()\n",
    "            if pred_token == tgt_vocab[EOS_token]:\n",
    "                break\n",
    "            output_labels.append(tgt_itos[pred_token])\n",
    "            input = torch.tensor([[pred_token]]).to(device)\n",
    "        \n",
    "        label = ' '.join(output_labels)\n",
    "        is_anomalous = \"anomalous\" in label.lower()\n",
    "        return label, is_anomalous\n",
    "\n",
    "# 训练模型\n",
    "train(model, dataloader, criterion, optimizer)\n",
    "\n",
    "# 测试异常检测\n",
    "print(\"\\n===== 测试异常检测 =====\")\n",
    "test_logs = [\n",
    "    \"[ERROR] Connection timeout occurred\",\n",
    "    \"[INFO] System started successfully\",\n",
    "    \"[WARN] High CPU usage 95%\",\n",
    "    \"[FATAL] Unknown error in module X\",\n",
    "]\n",
    "for log in test_logs:\n",
    "    label, is_anomalous = detect_anomaly(model, log, src_vocab, tgt_vocab, src_itos, tgt_itos)\n",
    "    print(f\"Log: {log} -> Predicted Label: {label} (Anomalous: {is_anomalous})\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a2dce515-fbb8-4237-a66c-1c71333dd495",
   "metadata": {},
   "source": [
    "### 示例4：日志分析与异常检测改进版\n",
    "\n",
    "- 扩展数据集为多条日志的序列\n",
    "- 使用自监督学习机制"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "10224cbb-5021-4a40-a257-9829f122d284",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import random\n",
    "from collections import Counter\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "import torch.nn.functional as F\n",
    "\n",
    "# 模拟日志数据集：每个样本是多条日志的序列\n",
    "log_sequences = [\n",
    "    [\"[INFO] System started successfully\", \"[WARN] Disk space low at 90%\", \"[ERROR] Connection timeout occurred\"],\n",
    "    [\"[INFO] User logged in\", \"[DEBUG] Request processed in 10ms\", \"[INFO] Backup completed\"],\n",
    "    [\"[FATAL] Database crash detected\", \"[ERROR] Invalid credentials provided\"],\n",
    "    [\"[INFO] All systems operational\", \"[INFO] Service restarted\"],\n",
    "    [\"[WARN] High CPU usage 95%\", \"[ERROR] Network failure\"],\n",
    "    [\"[INFO] System started successfully\", \"[INFO] User logged in\"],\n",
    "    [\"[ERROR] Connection timeout occurred\", \"[WARN] Disk space low at 90%\"],\n",
    "    [\"[DEBUG] Request processed in 10ms\", \"[INFO] Backup completed\"],\n",
    "    [\"[FATAL] Database crash detected\", \"[WARN] High CPU usage 95%\"],\n",
    "    [\"[INFO] All systems operational\", \"[ERROR] Invalid credentials provided\"],\n",
    "    [\"[INFO] Service restarted\", \"[INFO] User logged in\"],\n",
    "    [\"[ERROR] Network failure\", \"[WARN] Disk space low at 90%\"],\n",
    "]\n",
    "\n",
    "# 特殊标记\n",
    "SOS_token = \"<SOS>\"\n",
    "EOS_token = \"<EOS>\"\n",
    "PAD_token = \"<PAD>\"\n",
    "UNK_token = \"<UNK>\"\n",
    "SEQ_SEP = \"<SEP>\"\n",
    "\n",
    "# 构建词表\n",
    "def build_vocab(sequences):\n",
    "    counter = Counter()\n",
    "    for seq in sequences:\n",
    "        for log in seq:\n",
    "            words = log.split()\n",
    "            counter.update(words)\n",
    "    counter.update([SEQ_SEP])\n",
    "    vocab = {PAD_token: 0, SOS_token: 1, EOS_token: 2, UNK_token: 3}\n",
    "    vocab.update({word: i + 4 for i, word in enumerate(counter.keys())})\n",
    "    return vocab, {i: word for word, i in vocab.items()}\n",
    "\n",
    "vocab, itos = build_vocab(log_sequences)\n",
    "\n",
    "# 数据集类：自监督，目标序列 = 输入序列\n",
    "class LogDataset(Dataset):\n",
    "    def __init__(self, sequences, vocab):\n",
    "        self.sequences = sequences\n",
    "        self.vocab = vocab\n",
    "    \n",
    "    def __len__(self):\n",
    "        return len(self.sequences)\n",
    "    \n",
    "    def __getitem__(self, idx):\n",
    "        seq = self.sequences[idx]\n",
    "        flattened_seq = []\n",
    "        for log in seq:\n",
    "            flattened_seq.extend(log.split())\n",
    "            flattened_seq.append(SEQ_SEP)\n",
    "        flattened_seq = flattened_seq[:-1]  # 移除最后一个<SEP>\n",
    "        \n",
    "        indices = [self.vocab[SOS_token]] + [self.vocab.get(w, self.vocab[UNK_token]) for w in flattened_seq] + [self.vocab[EOS_token]]\n",
    "        return torch.tensor(indices), torch.tensor(indices)\n",
    "\n",
    "# 模型定义\n",
    "class Encoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Encoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size, hidden_size, batch_first=True)\n",
    "    \n",
    "    def forward(self, src):\n",
    "        embedded = self.embedding(src)\n",
    "        outputs, (hidden, cell) = self.lstm(embedded)\n",
    "        return outputs, hidden, cell\n",
    "\n",
    "class BahdanauAttention(nn.Module):\n",
    "    def __init__(self, hidden_size):\n",
    "        super(BahdanauAttention, self).__init__()\n",
    "        self.Wa = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Ua = nn.Linear(hidden_size, hidden_size)\n",
    "        self.Va = nn.Linear(hidden_size, 1)\n",
    "    \n",
    "    def forward(self, decoder_hidden, encoder_outputs):\n",
    "        score = self.Va(torch.tanh(self.Wa(decoder_hidden.unsqueeze(1)) + self.Ua(encoder_outputs)))\n",
    "        attn_weights = torch.softmax(score, dim=1)\n",
    "        context = torch.bmm(attn_weights.transpose(1, 2), encoder_outputs)\n",
    "        return context.squeeze(1), attn_weights.squeeze(2)\n",
    "\n",
    "class Decoder(nn.Module):\n",
    "    def __init__(self, vocab_size, embed_size, hidden_size):\n",
    "        super(Decoder, self).__init__()\n",
    "        self.embedding = nn.Embedding(vocab_size, embed_size)\n",
    "        self.lstm = nn.LSTM(embed_size + hidden_size, hidden_size, batch_first=True)\n",
    "        self.attention = BahdanauAttention(hidden_size)\n",
    "        self.out = nn.Linear(hidden_size, vocab_size)\n",
    "    \n",
    "    def forward(self, tgt, hidden, cell, encoder_outputs):\n",
    "        embedded = self.embedding(tgt)\n",
    "        context, attn_weights = self.attention(hidden[-1], encoder_outputs)\n",
    "        lstm_input = torch.cat((embedded, context.unsqueeze(1)), dim=2)\n",
    "        output, (hidden, cell) = self.lstm(lstm_input, (hidden, cell))\n",
    "        output = self.out(output.squeeze(1))\n",
    "        return output, hidden, cell, attn_weights\n",
    "\n",
    "class Seq2Seq(nn.Module):\n",
    "    def __init__(self, encoder, decoder):\n",
    "        super(Seq2Seq, self).__init__()\n",
    "        self.encoder = encoder\n",
    "        self.decoder = decoder\n",
    "    \n",
    "    def forward(self, src, tgt, teacher_forcing_ratio=0.5):\n",
    "        batch_size = src.size(0)\n",
    "        tgt_len = tgt.size(1)\n",
    "        tgt_vocab_size = self.decoder.out.out_features\n",
    "        \n",
    "        outputs = torch.zeros(batch_size, tgt_len, tgt_vocab_size).to(src.device)\n",
    "        encoder_outputs, hidden, cell = self.encoder(src)\n",
    "        \n",
    "        input = tgt[:, 0].unsqueeze(1)  # <SOS>\n",
    "        for t in range(1, tgt_len):\n",
    "            output, hidden, cell, _ = self.decoder(input, hidden, cell, encoder_outputs)\n",
    "            outputs[:, t, :] = output\n",
    "            teacher_force = random.random() < teacher_forcing_ratio\n",
    "            input = tgt[:, t].unsqueeze(1) if teacher_force else output.argmax(1).unsqueeze(1)\n",
    "        \n",
    "        return outputs\n",
    "\n",
    "# 数据准备\n",
    "dataset = LogDataset(log_sequences, vocab)\n",
    "dataloader = DataLoader(dataset, batch_size=2, shuffle=True, drop_last=True, collate_fn=lambda x: (\n",
    "    nn.utils.rnn.pad_sequence([item[0] for item in x], batch_first=True, padding_value=vocab[PAD_token]),\n",
    "    nn.utils.rnn.pad_sequence([item[1] for item in x], batch_first=True, padding_value=vocab[PAD_token])\n",
    "))\n",
    "\n",
    "# 模型参数\n",
    "embed_size = 128\n",
    "hidden_size = 256\n",
    "vocab_size = len(vocab)\n",
    "\n",
    "encoder = Encoder(vocab_size, embed_size, hidden_size)\n",
    "decoder = Decoder(vocab_size, embed_size, hidden_size)\n",
    "model = Seq2Seq(encoder, decoder)\n",
    "\n",
    "# 训练设置\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "model = model.to(device)\n",
    "criterion = nn.CrossEntropyLoss(ignore_index=vocab[PAD_token])\n",
    "optimizer = optim.Adam(model.parameters(), lr=0.001)\n",
    "\n",
    "# 训练循环\n",
    "def train(model, dataloader, criterion, optimizer, epochs=20):\n",
    "    model.train()\n",
    "    for epoch in range(epochs):\n",
    "        total_loss = 0\n",
    "        for src, tgt in dataloader:\n",
    "            src, tgt = src.to(device), tgt.to(device)\n",
    "            optimizer.zero_grad()\n",
    "            output = model(src, tgt)\n",
    "            loss = criterion(output[:, 1:, :].reshape(-1, vocab_size), tgt[:, 1:].reshape(-1))\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            total_loss += loss.item()\n",
    "        print(f'Epoch {epoch+1}, Loss: {total_loss / len(dataloader)}')\n",
    "\n",
    "# 异常检测函数（基于重建误差）\n",
    "def detect_anomaly(model, log_seq, vocab, itos, device, max_len=50, threshold=0.5):\n",
    "    model.eval()\n",
    "    with torch.no_grad():\n",
    "        # 扁平化输入序列\n",
    "        flattened_seq = []\n",
    "        for log in log_seq:\n",
    "            flattened_seq.extend(log.split())\n",
    "            flattened_seq.append(SEQ_SEP)\n",
    "        flattened_seq = flattened_seq[:-1]\n",
    "        \n",
    "        src_indices = [vocab[SOS_token]] + [vocab.get(w, vocab[UNK_token]) for w in flattened_seq] + [vocab[EOS_token]]\n",
    "        src_tensor = torch.tensor([src_indices], dtype=torch.long).to(device)\n",
    "        tgt_tensor = src_tensor.clone()  # 自监督，tgt = src\n",
    "        \n",
    "        # 使用模型forward生成重建输出（无Teacher Forcing，以模拟推理）\n",
    "        recon_outputs = model(src_tensor, tgt_tensor, teacher_forcing_ratio=0.0)\n",
    "        \n",
    "        # 计算重建误差\n",
    "        error = criterion(recon_outputs[:, 1:, :].reshape(-1, vocab_size), tgt_tensor[:, 1:].reshape(-1))\n",
    "        \n",
    "        # 生成重建序列以可视化\n",
    "        reconstructed = []\n",
    "        for t in range(1, recon_outputs.size(1)):\n",
    "            pred_token = recon_outputs[0, t].argmax(0).item()\n",
    "            if pred_token == vocab[EOS_token]:\n",
    "                break\n",
    "            reconstructed.append(itos[pred_token])\n",
    "        \n",
    "        is_anomalous = error.item() > threshold\n",
    "        reconstructed_seq = ' '.join(reconstructed).replace(SEQ_SEP, ' | ')\n",
    "        \n",
    "        return reconstructed_seq, error.item(), is_anomalous\n",
    "\n",
    "# 训练模型\n",
    "train(model, dataloader, criterion, optimizer)\n",
    "\n",
    "# 测试异常检测\n",
    "print(\"\\n===== 测试异常检测 =====\")\n",
    "test_sequences = [\n",
    "    [\"[INFO] System started successfully\", \"[WARN] Disk space low at 90%\", \"[ERROR] Connection timeout occurred\"],\n",
    "    [\"[INFO] User logged in\", \"[DEBUG] Request processed in 10ms\"],\n",
    "    [\"[FATAL] Unknown error detected\", \"[ERROR] System overload\"],\n",
    "]\n",
    "for seq in test_sequences:\n",
    "    reconstructed, error, is_anomalous = detect_anomaly(model, seq, vocab, itos, device)\n",
    "    print(f\"Input Sequence: {' | '.join(seq)}\\nReconstructed: {reconstructed}\\nError: {error:.4f} (Anomalous: {is_anomalous})\\n\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
